aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorTill <2353100+S7evinK@users.noreply.github.com>2023-03-22 09:21:32 +0100
committerGitHub <noreply@github.com>2023-03-22 09:21:32 +0100
commit5e85a00cb36c3d343cd5b6f6a18435989724a135 (patch)
treeb22c34dd0a6cdc04025b90086843f9084a876412
parentec6879e5ae2919c903707475ce8d72244b2a6847 (diff)
Remove `BaseDendrite` (#3023)
Removes `BaseDendrite` to, hopefully, make testing and composing of components easier in the future.
-rw-r--r--appservice/appservice.go21
-rw-r--r--appservice/appservice_test.go23
-rw-r--r--build/dendritejs-pinecone/main.go40
-rw-r--r--build/gobind-pinecone/monolith.go13
-rw-r--r--build/gobind-yggdrasil/monolith.go89
-rw-r--r--clientapi/admin_test.go50
-rw-r--r--clientapi/clientapi.go29
-rw-r--r--clientapi/routing/auth_fallback_test.go46
-rw-r--r--clientapi/routing/joinroom_test.go20
-rw-r--r--clientapi/routing/login_test.go25
-rw-r--r--clientapi/routing/register_test.go74
-rw-r--r--clientapi/routing/routing.go23
-rw-r--r--cmd/dendrite-demo-pinecone/conn/client.go8
-rw-r--r--cmd/dendrite-demo-pinecone/main.go9
-rw-r--r--cmd/dendrite-demo-pinecone/monolith/monolith.go81
-rw-r--r--cmd/dendrite-demo-yggdrasil/main.go94
-rw-r--r--cmd/dendrite-demo-yggdrasil/yggconn/client.go10
-rw-r--r--cmd/dendrite-upgrade-tests/main.go16
-rw-r--r--cmd/dendrite/main.go111
-rw-r--r--cmd/resolve-state/main.go9
-rw-r--r--federationapi/federationapi.go51
-rw-r--r--federationapi/federationapi_keys_test.go11
-rw-r--r--federationapi/federationapi_test.go75
-rw-r--r--federationapi/queue/queue_test.go14
-rw-r--r--federationapi/routing/profile_test.go18
-rw-r--r--federationapi/routing/query_test.go18
-rw-r--r--federationapi/routing/routing.go16
-rw-r--r--federationapi/routing/send_test.go18
-rw-r--r--federationapi/storage/storage_test.go2
-rw-r--r--internal/sqlutil/connection_manager.go31
-rw-r--r--internal/sqlutil/connection_manager_test.go8
-rw-r--r--mediaapi/mediaapi.go15
-rw-r--r--mediaapi/routing/routing.go15
-rw-r--r--mediaapi/routing/upload_test.go2
-rw-r--r--mediaapi/storage/storage_test.go2
-rw-r--r--relayapi/relayapi.go23
-rw-r--r--relayapi/relayapi_test.go65
-rw-r--r--roomserver/internal/alias.go8
-rw-r--r--roomserver/internal/api.go50
-rw-r--r--roomserver/internal/helpers/helpers_test.go2
-rw-r--r--roomserver/internal/input/input.go9
-rw-r--r--roomserver/internal/input/input_test.go117
-rw-r--r--roomserver/roomserver.go19
-rw-r--r--roomserver/roomserver_test.go71
-rw-r--r--roomserver/storage/shared/storage_test.go4
-rw-r--r--setup/base/base.go242
-rw-r--r--setup/base/base_test.go20
-rw-r--r--setup/base/sanity_other.go2
-rw-r--r--setup/base/sanity_unix.go2
-rw-r--r--setup/jetstream/nats.go4
-rw-r--r--setup/monolith.go27
-rw-r--r--setup/mscs/msc2836/msc2836.go13
-rw-r--r--setup/mscs/msc2836/msc2836_test.go15
-rw-r--r--setup/mscs/msc2946/msc2946.go18
-rw-r--r--setup/mscs/mscs.go16
-rw-r--r--syncapi/storage/storage_test.go2
-rw-r--r--syncapi/syncapi.go48
-rw-r--r--syncapi/syncapi_test.go162
-rw-r--r--test/testrig/base.go43
-rw-r--r--test/testrig/jetstream.go6
-rw-r--r--userapi/consumers/roomserver_test.go2
-rw-r--r--userapi/internal/device_list_update_test.go2
-rw-r--r--userapi/internal/key_api_test.go2
-rw-r--r--userapi/storage/storage_test.go7
-rw-r--r--userapi/userapi.go58
-rw-r--r--userapi/userapi_test.go26
-rw-r--r--userapi/util/notify_test.go2
-rw-r--r--userapi/util/phonehomestats_test.go14
68 files changed, 1186 insertions, 1002 deletions
diff --git a/appservice/appservice.go b/appservice/appservice.go
index 5b1b93de..d13d9eb1 100644
--- a/appservice/appservice.go
+++ b/appservice/appservice.go
@@ -21,6 +21,8 @@ import (
"sync"
"time"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/sirupsen/logrus"
"github.com/matrix-org/gomatrixserverlib"
@@ -29,7 +31,6 @@ import (
"github.com/matrix-org/dendrite/appservice/consumers"
"github.com/matrix-org/dendrite/appservice/query"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
)
@@ -37,7 +38,9 @@ import (
// NewInternalAPI returns a concerete implementation of the internal API. Callers
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
func NewInternalAPI(
- base *base.BaseDendrite,
+ processContext *process.ProcessContext,
+ cfg *config.Dendrite,
+ natsInstance *jetstream.NATSInstance,
userAPI userapi.AppserviceUserAPI,
rsAPI roomserverAPI.RoomserverInternalAPI,
) appserviceAPI.AppServiceInternalAPI {
@@ -46,7 +49,7 @@ func NewInternalAPI(
Transport: &http.Transport{
DisableKeepAlives: true,
TLSClientConfig: &tls.Config{
- InsecureSkipVerify: base.Cfg.AppServiceAPI.DisableTLSValidation,
+ InsecureSkipVerify: cfg.AppServiceAPI.DisableTLSValidation,
},
Proxy: http.ProxyFromEnvironment,
},
@@ -55,21 +58,21 @@ func NewInternalAPI(
// outbound and inbound requests (inbound only for the internal API)
appserviceQueryAPI := &query.AppServiceQueryAPI{
HTTPClient: client,
- Cfg: &base.Cfg.AppServiceAPI,
+ Cfg: &cfg.AppServiceAPI,
ProtocolCache: map[string]appserviceAPI.ASProtocolResponse{},
CacheMu: sync.Mutex{},
}
- if len(base.Cfg.Derived.ApplicationServices) == 0 {
+ if len(cfg.Derived.ApplicationServices) == 0 {
return appserviceQueryAPI
}
// Wrap application services in a type that relates the application service and
// a sync.Cond object that can be used to notify workers when there are new
// events to be sent out.
- for _, appservice := range base.Cfg.Derived.ApplicationServices {
+ for _, appservice := range cfg.Derived.ApplicationServices {
// Create bot account for this AS if it doesn't already exist
- if err := generateAppServiceAccount(userAPI, appservice, base.Cfg.Global.ServerName); err != nil {
+ if err := generateAppServiceAccount(userAPI, appservice, cfg.Global.ServerName); err != nil {
logrus.WithFields(logrus.Fields{
"appservice": appservice.ID,
}).WithError(err).Panicf("failed to generate bot account for appservice")
@@ -78,9 +81,9 @@ func NewInternalAPI(
// Only consume if we actually have ASes to track, else we'll just chew cycles needlessly.
// We can't add ASes at runtime so this is safe to do.
- js, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
+ js, _ := natsInstance.Prepare(processContext, &cfg.Global.JetStream)
consumer := consumers.NewOutputRoomEventConsumer(
- base.ProcessContext, &base.Cfg.AppServiceAPI,
+ processContext, &cfg.AppServiceAPI,
client, js, rsAPI,
)
if err := consumer.Start(); err != nil {
diff --git a/appservice/appservice_test.go b/appservice/appservice_test.go
index ad6f1dfc..6c8a07b5 100644
--- a/appservice/appservice_test.go
+++ b/appservice/appservice_test.go
@@ -9,12 +9,15 @@ import (
"regexp"
"strings"
"testing"
+ "time"
"github.com/matrix-org/dendrite/appservice"
"github.com/matrix-org/dendrite/appservice/api"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/userapi"
@@ -105,11 +108,11 @@ func TestAppserviceInternalAPI(t *testing.T) {
}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, closeBase := testrig.CreateBaseDendrite(t, dbType)
- defer closeBase()
+ cfg, ctx, close := testrig.CreateConfig(t, dbType)
+ defer close()
// Create a dummy application service
- base.Cfg.AppServiceAPI.Derived.ApplicationServices = []config.ApplicationService{
+ cfg.AppServiceAPI.Derived.ApplicationServices = []config.ApplicationService{
{
ID: "someID",
URL: srv.URL,
@@ -124,11 +127,17 @@ func TestAppserviceInternalAPI(t *testing.T) {
},
}
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
+ t.Cleanup(func() {
+ ctx.ShutdownDendrite()
+ ctx.WaitForShutdown()
+ })
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
// Create required internal APIs
- rsAPI := roomserver.NewInternalAPI(base, caches)
- usrAPI := userapi.NewInternalAPI(base, rsAPI, nil)
- asAPI := appservice.NewInternalAPI(base, usrAPI, rsAPI)
+ natsInstance := jetstream.NATSInstance{}
+ cm := sqlutil.NewConnectionManager(ctx, cfg.Global.DatabaseOptions)
+ rsAPI := roomserver.NewInternalAPI(ctx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
+ usrAPI := userapi.NewInternalAPI(ctx, cfg, cm, &natsInstance, rsAPI, nil)
+ asAPI := appservice.NewInternalAPI(ctx, cfg, &natsInstance, usrAPI, rsAPI)
runCases(t, asAPI)
})
diff --git a/build/dendritejs-pinecone/main.go b/build/dendritejs-pinecone/main.go
index 96f034bd..bc9535fc 100644
--- a/build/dendritejs-pinecone/main.go
+++ b/build/dendritejs-pinecone/main.go
@@ -31,10 +31,12 @@ import (
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/userapi"
"github.com/matrix-org/gomatrixserverlib"
@@ -158,9 +160,8 @@ func startup() {
pManager.AddPeer("wss://pinecone.matrix.org/public")
cfg := &config.Dendrite{}
- cfg.Defaults(true)
+ cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: false})
cfg.UserAPI.AccountDatabase.ConnectionString = "file:/idb/dendritejs_account.db"
- cfg.AppServiceAPI.Database.ConnectionString = "file:/idb/dendritejs_appservice.db"
cfg.FederationAPI.Database.ConnectionString = "file:/idb/dendritejs_fedsender.db"
cfg.MediaAPI.Database.ConnectionString = "file:/idb/dendritejs_mediaapi.db"
cfg.RoomServer.Database.ConnectionString = "file:/idb/dendritejs_roomserver.db"
@@ -177,29 +178,30 @@ func startup() {
if err := cfg.Derive(); err != nil {
logrus.Fatalf("Failed to derive values from config: %s", err)
}
- base := base.NewBaseDendrite(cfg)
- defer base.Close() // nolint: errcheck
+ natsInstance := jetstream.NATSInstance{}
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+ caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
- rsAPI := roomserver.NewInternalAPI(base)
-
- federation := conn.CreateFederationClient(base, pSessions)
+ federation := conn.CreateFederationClient(cfg, pSessions)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
- userAPI := userapi.NewInternalAPI(base, rsAPI, federation)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
asQuery := appservice.NewInternalAPI(
- base, userAPI, rsAPI,
+ processCtx, cfg, &natsInstance, userAPI, rsAPI,
)
rsAPI.SetAppserviceAPI(asQuery)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.EnableMetrics)
- fedSenderAPI := federationapi.NewInternalAPI(base, federation, rsAPI, caches, keyRing, true)
+ fedSenderAPI := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true)
rsAPI.SetFederationAPI(fedSenderAPI, keyRing)
monolith := setup.Monolith{
- Config: base.Cfg,
- Client: conn.CreateClient(base, pSessions),
+ Config: cfg,
+ Client: conn.CreateClient(pSessions),
FedClient: federation,
KeyRing: keyRing,
@@ -210,15 +212,15 @@ func startup() {
//ServerKeyAPI: serverKeyAPI,
ExtPublicRoomsProvider: rooms.NewPineconeRoomProvider(pRouter, pSessions, fedSenderAPI, federation),
}
- monolith.AddAllPublicRoutes(base, caches)
+ monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
httpRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
- httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(base.Routers.Client)
- httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
+ httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
+ httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
p2pRouter := pSessions.Protocol("matrix").HTTP().Mux()
- p2pRouter.Handle(httputil.PublicFederationPathPrefix, base.Routers.Federation)
- p2pRouter.Handle(httputil.PublicMediaPathPrefix, base.Routers.Media)
+ p2pRouter.Handle(httputil.PublicFederationPathPrefix, routers.Federation)
+ p2pRouter.Handle(httputil.PublicMediaPathPrefix, routers.Media)
// Expose the matrix APIs via fetch - for local traffic
go func() {
diff --git a/build/gobind-pinecone/monolith.go b/build/gobind-pinecone/monolith.go
index 16797eec..2e2ca04d 100644
--- a/build/gobind-pinecone/monolith.go
+++ b/build/gobind-pinecone/monolith.go
@@ -30,6 +30,9 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-pinecone/relay"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/federationapi/api"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/process"
userapiAPI "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/pinecone/types"
@@ -187,7 +190,7 @@ func (m *DendriteMonolith) SetRelayServers(nodeID string, uris string) {
relay.UpdateNodeRelayServers(
gomatrixserverlib.ServerName(nodeKey),
relays,
- m.p2pMonolith.BaseDendrite.Context(),
+ m.p2pMonolith.ProcessCtx.Context(),
m.p2pMonolith.GetFederationAPI(),
)
}
@@ -214,7 +217,7 @@ func (m *DendriteMonolith) GetRelayServers(nodeID string) string {
} else {
request := api.P2PQueryRelayServersRequest{Server: gomatrixserverlib.ServerName(nodeKey)}
response := api.P2PQueryRelayServersResponse{}
- err := m.p2pMonolith.GetFederationAPI().P2PQueryRelayServers(m.p2pMonolith.BaseDendrite.Context(), &request, &response)
+ err := m.p2pMonolith.GetFederationAPI().P2PQueryRelayServers(m.p2pMonolith.ProcessCtx.Context(), &request, &response)
if err != nil {
logrus.Warnf("Failed obtaining list of this node's relay servers: %s", err.Error())
return ""
@@ -346,10 +349,14 @@ func (m *DendriteMonolith) Start() {
// This isn't actually fixed: https://github.com/blevesearch/zapx/pull/147
cfg.SyncAPI.Fulltext.Enabled = false
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+
enableRelaying := false
enableMetrics := false
enableWebsockets := false
- m.p2pMonolith.SetupDendrite(cfg, 65432, enableRelaying, enableMetrics, enableWebsockets)
+ m.p2pMonolith.SetupDendrite(processCtx, cfg, cm, routers, 65432, enableRelaying, enableMetrics, enableWebsockets)
m.p2pMonolith.StartMonolith()
}
diff --git a/build/gobind-yggdrasil/monolith.go b/build/gobind-yggdrasil/monolith.go
index 8faad1d0..7ce1892c 100644
--- a/build/gobind-yggdrasil/monolith.go
+++ b/build/gobind-yggdrasil/monolith.go
@@ -12,6 +12,7 @@ import (
"path/filepath"
"time"
+ "github.com/getsentry/sentry-go"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/appservice"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
@@ -19,12 +20,15 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/yggrooms"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/api"
+ "github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup"
- "github.com/matrix-org/dendrite/setup/base"
+ basepkg "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/userapi"
@@ -149,26 +153,71 @@ func (m *DendriteMonolith) Start() {
panic(err)
}
- base := base.NewBaseDendrite(cfg)
- base.ConfigureAdminEndpoints()
- m.processContext = base.ProcessContext
- defer base.Close() // nolint: errcheck
+ configErrors := &config.ConfigErrors{}
+ cfg.Verify(configErrors)
+ if len(*configErrors) > 0 {
+ for _, err := range *configErrors {
+ logrus.Errorf("Configuration error: %s", err)
+ }
+ logrus.Fatalf("Failed to start due to configuration errors")
+ }
+
+ internal.SetupStdLogging()
+ internal.SetupHookLogging(cfg.Logging)
+ internal.SetupPprof()
+
+ logrus.Infof("Dendrite version %s", internal.VersionString())
+
+ if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
+ logrus.Warn("Open registration is enabled")
+ }
- federation := ygg.CreateFederationClient(base)
+ closer, err := cfg.SetupTracing()
+ if err != nil {
+ logrus.WithError(err).Panicf("failed to start opentracing")
+ }
+ defer closer.Close()
+
+ if cfg.Global.Sentry.Enabled {
+ logrus.Info("Setting up Sentry for debugging...")
+ err = sentry.Init(sentry.ClientOptions{
+ Dsn: cfg.Global.Sentry.DSN,
+ Environment: cfg.Global.Sentry.Environment,
+ Debug: true,
+ ServerName: string(cfg.Global.ServerName),
+ Release: "dendrite@" + internal.VersionString(),
+ AttachStacktrace: true,
+ })
+ if err != nil {
+ logrus.WithError(err).Panic("failed to start Sentry")
+ }
+ }
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+ basepkg.ConfigureAdminEndpoints(processCtx, routers)
+ m.processContext = processCtx
+ defer func() {
+ processCtx.ShutdownDendrite()
+ processCtx.WaitForShutdown()
+ }() // nolint: errcheck
+
+ federation := ygg.CreateFederationClient(cfg)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
- caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
fsAPI := federationapi.NewInternalAPI(
- base, federation, rsAPI, caches, keyRing, true,
+ processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
- userAPI := userapi.NewInternalAPI(base, rsAPI, federation)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
- asAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
+ asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
// The underlying roomserver implementation needs to be able to call the fedsender.
@@ -176,8 +225,8 @@ func (m *DendriteMonolith) Start() {
rsAPI.SetFederationAPI(fsAPI, keyRing)
monolith := setup.Monolith{
- Config: base.Cfg,
- Client: ygg.CreateClient(base),
+ Config: cfg,
+ Client: ygg.CreateClient(),
FedClient: federation,
KeyRing: keyRing,
@@ -189,17 +238,17 @@ func (m *DendriteMonolith) Start() {
ygg, fsAPI, federation,
),
}
- monolith.AddAllPublicRoutes(base, caches)
+ monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
httpRouter := mux.NewRouter()
- httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(base.Routers.Client)
- httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
- httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(base.Routers.DendriteAdmin)
- httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(base.Routers.SynapseAdmin)
+ httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
+ httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
+ httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
+ httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
yggRouter := mux.NewRouter()
- yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(base.Routers.Federation)
- yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
+ yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(routers.Federation)
+ yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
// Build both ends of a HTTP multiplex.
m.httpServer = &http.Server{
diff --git a/clientapi/admin_test.go b/clientapi/admin_test.go
index 46e2d303..4d2bf67b 100644
--- a/clientapi/admin_test.go
+++ b/clientapi/admin_test.go
@@ -5,13 +5,17 @@ import (
"net/http"
"net/http/httptest"
"testing"
+ "time"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/syncapi"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@@ -30,20 +34,22 @@ func TestAdminResetPassword(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
-
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ defer close()
+ natsInstance := jetstream.NATSInstance{}
// add a vhost
- base.Cfg.Global.VirtualHosts = append(base.Cfg.Global.VirtualHosts, &config.VirtualHost{
+ cfg.Global.VirtualHosts = append(cfg.Global.VirtualHosts, &config.VirtualHost{
SigningIdentity: gomatrixserverlib.SigningIdentity{ServerName: "vh1"},
})
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
// Needed for changing the password/login
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the userAPI for this test, so nil for other APIs/caches etc.
- AddPublicRoutes(base, nil, rsAPI, nil, nil, nil, userAPI, nil, nil)
+ AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
// Create the users in the userapi and login
accessTokens := map[*test.User]string{
@@ -73,7 +79,7 @@ func TestAdminResetPassword(t *testing.T) {
"password": password,
}))
rec := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(rec, req)
+ routers.Client.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("failed to login: %s", rec.Body.String())
}
@@ -126,7 +132,7 @@ func TestAdminResetPassword(t *testing.T) {
}
rec := httptest.NewRecorder()
- base.Routers.DendriteAdmin.ServeHTTP(rec, req)
+ routers.DendriteAdmin.ServeHTTP(rec, req)
t.Logf("%s", rec.Body.String())
if tc.wantOK && rec.Code != http.StatusOK {
t.Fatalf("expected http status %d, got %d: %s", http.StatusOK, rec.Code, rec.Body.String())
@@ -149,17 +155,19 @@ func TestPurgeRoom(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- defer baseClose()
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ defer close()
- fedClient := base.CreateFederationClient()
- rsAPI := roomserver.NewInternalAPI(base, caches)
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// this starts the JetStream consumers
- syncapi.AddPublicRoutes(base, userAPI, rsAPI, caches)
- federationapi.NewInternalAPI(base, fedClient, rsAPI, caches, nil, true)
+ syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, userAPI, rsAPI, caches, caching.DisableMetrics)
+ federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(nil, nil)
// Create the room
@@ -168,7 +176,7 @@ func TestPurgeRoom(t *testing.T) {
}
// We mostly need the rsAPI for this test, so nil for other APIs/caches etc.
- AddPublicRoutes(base, nil, rsAPI, nil, nil, nil, userAPI, nil, nil)
+ AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, rsAPI, nil, nil, nil, userAPI, nil, nil, caching.DisableMetrics)
// Create the users in the userapi and login
accessTokens := map[*test.User]string{
@@ -196,7 +204,7 @@ func TestPurgeRoom(t *testing.T) {
"password": password,
}))
rec := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(rec, req)
+ routers.Client.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("failed to login: %s", rec.Body.String())
}
@@ -221,7 +229,7 @@ func TestPurgeRoom(t *testing.T) {
req.Header.Set("Authorization", "Bearer "+accessTokens[aliceAdmin])
rec := httptest.NewRecorder()
- base.Routers.DendriteAdmin.ServeHTTP(rec, req)
+ routers.DendriteAdmin.ServeHTTP(rec, req)
t.Logf("%s", rec.Body.String())
if tc.wantOK && rec.Code != http.StatusOK {
t.Fatalf("expected http status %d, got %d: %s", http.StatusOK, rec.Code, rec.Body.String())
diff --git a/clientapi/clientapi.go b/clientapi/clientapi.go
index e9985d43..d35f1d4d 100644
--- a/clientapi/clientapi.go
+++ b/clientapi/clientapi.go
@@ -15,6 +15,9 @@
package clientapi
import (
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
@@ -25,13 +28,15 @@ import (
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/transactions"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
)
// AddPublicRoutes sets up and registers HTTP handlers for the ClientAPI component.
func AddPublicRoutes(
- base *base.BaseDendrite,
+ processContext *process.ProcessContext,
+ routers httputil.Routers,
+ cfg *config.Dendrite,
+ natsInstance *jetstream.NATSInstance,
federation *gomatrixserverlib.FederationClient,
rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
@@ -39,27 +44,25 @@ func AddPublicRoutes(
fsAPI federationAPI.ClientFederationAPI,
userAPI userapi.ClientUserAPI,
userDirectoryProvider userapi.QuerySearchProfilesAPI,
- extRoomsProvider api.ExtraPublicRoomsProvider,
+ extRoomsProvider api.ExtraPublicRoomsProvider, enableMetrics bool,
) {
- cfg := &base.Cfg.ClientAPI
- mscCfg := &base.Cfg.MSCs
- js, natsClient := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
+ js, natsClient := natsInstance.Prepare(processContext, &cfg.Global.JetStream)
syncProducer := &producers.SyncAPIProducer{
JetStream: js,
- TopicReceiptEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
- TopicSendToDeviceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
- TopicTypingEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
- TopicPresenceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
+ TopicReceiptEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputReceiptEvent),
+ TopicSendToDeviceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
+ TopicTypingEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputTypingEvent),
+ TopicPresenceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
UserAPI: userAPI,
- ServerName: cfg.Matrix.ServerName,
+ ServerName: cfg.Global.ServerName,
}
routing.Setup(
- base,
+ routers,
cfg, rsAPI, asAPI,
userAPI, userDirectoryProvider, federation,
syncProducer, transactionsCache, fsAPI,
- extRoomsProvider, mscCfg, natsClient,
+ extRoomsProvider, natsClient, enableMetrics,
)
}
diff --git a/clientapi/routing/auth_fallback_test.go b/clientapi/routing/auth_fallback_test.go
index 534581bd..afeca051 100644
--- a/clientapi/routing/auth_fallback_test.go
+++ b/clientapi/routing/auth_fallback_test.go
@@ -10,30 +10,28 @@ import (
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/setup/config"
- "github.com/matrix-org/dendrite/test/testrig"
)
func Test_AuthFallback(t *testing.T) {
- base, _, _ := testrig.Base(nil)
- defer base.Close()
-
+ cfg := config.Dendrite{}
+ cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
for _, useHCaptcha := range []bool{false, true} {
for _, recaptchaEnabled := range []bool{false, true} {
for _, wantErr := range []bool{false, true} {
t.Run(fmt.Sprintf("useHCaptcha(%v) - recaptchaEnabled(%v) - wantErr(%v)", useHCaptcha, recaptchaEnabled, wantErr), func(t *testing.T) {
// Set the defaults for each test
- base.Cfg.ClientAPI.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
- base.Cfg.ClientAPI.RecaptchaEnabled = recaptchaEnabled
- base.Cfg.ClientAPI.RecaptchaPublicKey = "pub"
- base.Cfg.ClientAPI.RecaptchaPrivateKey = "priv"
+ cfg.ClientAPI.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
+ cfg.ClientAPI.RecaptchaEnabled = recaptchaEnabled
+ cfg.ClientAPI.RecaptchaPublicKey = "pub"
+ cfg.ClientAPI.RecaptchaPrivateKey = "priv"
if useHCaptcha {
- base.Cfg.ClientAPI.RecaptchaSiteVerifyAPI = "https://hcaptcha.com/siteverify"
- base.Cfg.ClientAPI.RecaptchaApiJsUrl = "https://js.hcaptcha.com/1/api.js"
- base.Cfg.ClientAPI.RecaptchaFormField = "h-captcha-response"
- base.Cfg.ClientAPI.RecaptchaSitekeyClass = "h-captcha"
+ cfg.ClientAPI.RecaptchaSiteVerifyAPI = "https://hcaptcha.com/siteverify"
+ cfg.ClientAPI.RecaptchaApiJsUrl = "https://js.hcaptcha.com/1/api.js"
+ cfg.ClientAPI.RecaptchaFormField = "h-captcha-response"
+ cfg.ClientAPI.RecaptchaSitekeyClass = "h-captcha"
}
cfgErrs := &config.ConfigErrors{}
- base.Cfg.ClientAPI.Verify(cfgErrs)
+ cfg.ClientAPI.Verify(cfgErrs)
if len(*cfgErrs) > 0 {
t.Fatalf("(hCaptcha=%v) unexpected config errors: %s", useHCaptcha, cfgErrs.Error())
}
@@ -41,7 +39,7 @@ func Test_AuthFallback(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/?session=1337", nil)
rec := httptest.NewRecorder()
- AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
+ AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if !recaptchaEnabled {
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected response code: %d, want %d", rec.Code, http.StatusBadRequest)
@@ -50,8 +48,8 @@ func Test_AuthFallback(t *testing.T) {
t.Fatalf("unexpected response body: %s", rec.Body.String())
}
} else {
- if !strings.Contains(rec.Body.String(), base.Cfg.ClientAPI.RecaptchaSitekeyClass) {
- t.Fatalf("body does not contain %s: %s", base.Cfg.ClientAPI.RecaptchaSitekeyClass, rec.Body.String())
+ if !strings.Contains(rec.Body.String(), cfg.ClientAPI.RecaptchaSitekeyClass) {
+ t.Fatalf("body does not contain %s: %s", cfg.ClientAPI.RecaptchaSitekeyClass, rec.Body.String())
}
}
@@ -64,14 +62,14 @@ func Test_AuthFallback(t *testing.T) {
}))
defer srv.Close() // nolint: errcheck
- base.Cfg.ClientAPI.RecaptchaSiteVerifyAPI = srv.URL
+ cfg.ClientAPI.RecaptchaSiteVerifyAPI = srv.URL
// check the result after sending the captcha
req = httptest.NewRequest(http.MethodPost, "/?session=1337", nil)
req.Form = url.Values{}
- req.Form.Add(base.Cfg.ClientAPI.RecaptchaFormField, "someRandomValue")
+ req.Form.Add(cfg.ClientAPI.RecaptchaFormField, "someRandomValue")
rec = httptest.NewRecorder()
- AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
+ AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if recaptchaEnabled {
if !wantErr {
if rec.Code != http.StatusOK {
@@ -105,7 +103,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("unknown fallbacks are handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/?session=1337", nil)
rec := httptest.NewRecorder()
- AuthFallback(rec, req, "DoesNotExist", &base.Cfg.ClientAPI)
+ AuthFallback(rec, req, "DoesNotExist", &cfg.ClientAPI)
if rec.Code != http.StatusNotImplemented {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusNotImplemented)
}
@@ -114,7 +112,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("unknown methods are handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodDelete, "/?session=1337", nil)
rec := httptest.NewRecorder()
- AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
+ AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusMethodNotAllowed {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusMethodNotAllowed)
}
@@ -123,7 +121,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("missing session parameter is handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/", nil)
rec := httptest.NewRecorder()
- AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
+ AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusBadRequest)
}
@@ -132,7 +130,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("missing session parameter is handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/", nil)
rec := httptest.NewRecorder()
- AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
+ AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusBadRequest)
}
@@ -141,7 +139,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("missing 'response' is handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/?session=1337", nil)
rec := httptest.NewRecorder()
- AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
+ AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusBadRequest)
}
diff --git a/clientapi/routing/joinroom_test.go b/clientapi/routing/joinroom_test.go
index de8f9538..fd58ff5d 100644
--- a/clientapi/routing/joinroom_test.go
+++ b/clientapi/routing/joinroom_test.go
@@ -8,6 +8,8 @@ import (
"time"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/appservice"
@@ -25,13 +27,15 @@ func TestJoinRoomByIDOrAlias(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ defer close()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
- asAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
+ asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetFederationAPI(nil, nil) // creates the rs.Inputer etc
// Create the users in the userapi
@@ -63,7 +67,7 @@ func TestJoinRoomByIDOrAlias(t *testing.T) {
RoomAliasName: "alias",
Invite: []string{bob.ID},
GuestCanJoin: false,
- }, aliceDev, &base.Cfg.ClientAPI, userAPI, rsAPI, asAPI, time.Now())
+ }, aliceDev, &cfg.ClientAPI, userAPI, rsAPI, asAPI, time.Now())
crResp, ok := resp.JSON.(createRoomResponse)
if !ok {
t.Fatalf("response is not a createRoomResponse: %+v", resp)
@@ -78,7 +82,7 @@ func TestJoinRoomByIDOrAlias(t *testing.T) {
Preset: presetPublicChat,
Invite: []string{charlie.ID},
GuestCanJoin: true,
- }, aliceDev, &base.Cfg.ClientAPI, userAPI, rsAPI, asAPI, time.Now())
+ }, aliceDev, &cfg.ClientAPI, userAPI, rsAPI, asAPI, time.Now())
crRespWithGuestAccess, ok := resp.JSON.(createRoomResponse)
if !ok {
t.Fatalf("response is not a createRoomResponse: %+v", resp)
diff --git a/clientapi/routing/login_test.go b/clientapi/routing/login_test.go
index fd3d8cba..b2773076 100644
--- a/clientapi/routing/login_test.go
+++ b/clientapi/routing/login_test.go
@@ -7,11 +7,15 @@ import (
"net/http/httptest"
"strings"
"testing"
+ "time"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@@ -29,21 +33,24 @@ func TestLogin(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
- base.Cfg.ClientAPI.RateLimiting.Enabled = false
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ defer close()
+ cfg.ClientAPI.RateLimiting.Enabled = false
+ natsInstance := jetstream.NATSInstance{}
// add a vhost
- base.Cfg.Global.VirtualHosts = append(base.Cfg.Global.VirtualHosts, &config.VirtualHost{
+ cfg.Global.VirtualHosts = append(cfg.Global.VirtualHosts, &config.VirtualHost{
SigningIdentity: gomatrixserverlib.SigningIdentity{ServerName: "vh1"},
})
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
// Needed for /login
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// We mostly need the userAPI for this test, so nil for other APIs/caches etc.
- Setup(base, &base.Cfg.ClientAPI, nil, nil, userAPI, nil, nil, nil, nil, nil, nil, &base.Cfg.MSCs, nil)
+ Setup(routers, cfg, nil, nil, userAPI, nil, nil, nil, nil, nil, nil, nil, caching.DisableMetrics)
// Create password
password := util.RandomString(8)
@@ -116,7 +123,7 @@ func TestLogin(t *testing.T) {
"password": password,
}))
rec := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(rec, req)
+ routers.Client.ServeHTTP(rec, req)
if tc.wantOK && rec.Code != http.StatusOK {
t.Fatalf("failed to login: %s", rec.Body.String())
}
diff --git a/clientapi/routing/register_test.go b/clientapi/routing/register_test.go
index c06b0ae1..46cd8b2b 100644
--- a/clientapi/routing/register_test.go
+++ b/clientapi/routing/register_test.go
@@ -31,8 +31,10 @@ import (
"github.com/matrix-org/dendrite/clientapi/jsonerror"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/dendrite/userapi"
@@ -405,12 +407,15 @@ func Test_register(t *testing.T) {
}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ defer close()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
@@ -432,16 +437,16 @@ func Test_register(t *testing.T) {
}
}))
defer srv.Close()
- base.Cfg.ClientAPI.RecaptchaSiteVerifyAPI = srv.URL
+ cfg.ClientAPI.RecaptchaSiteVerifyAPI = srv.URL
}
- if err := base.Cfg.Derive(); err != nil {
+ if err := cfg.Derive(); err != nil {
t.Fatalf("failed to derive config: %s", err)
}
- base.Cfg.ClientAPI.RecaptchaEnabled = tc.enableRecaptcha
- base.Cfg.ClientAPI.RegistrationDisabled = tc.registrationDisabled
- base.Cfg.ClientAPI.GuestsDisabled = tc.guestsDisabled
+ cfg.ClientAPI.RecaptchaEnabled = tc.enableRecaptcha
+ cfg.ClientAPI.RegistrationDisabled = tc.registrationDisabled
+ cfg.ClientAPI.GuestsDisabled = tc.guestsDisabled
if tc.kind == "" {
tc.kind = "user"
@@ -469,15 +474,15 @@ func Test_register(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/?kind=%s", tc.kind), body)
- resp := Register(req, userAPI, &base.Cfg.ClientAPI)
+ resp := Register(req, userAPI, &cfg.ClientAPI)
t.Logf("Resp: %+v", resp)
// The first request should return a userInteractiveResponse
switch r := resp.JSON.(type) {
case userInteractiveResponse:
// Check that the flows are the ones we configured
- if !reflect.DeepEqual(r.Flows, base.Cfg.Derived.Registration.Flows) {
- t.Fatalf("unexpected registration flows: %+v, want %+v", r.Flows, base.Cfg.Derived.Registration.Flows)
+ if !reflect.DeepEqual(r.Flows, cfg.Derived.Registration.Flows) {
+ t.Fatalf("unexpected registration flows: %+v, want %+v", r.Flows, cfg.Derived.Registration.Flows)
}
case *jsonerror.MatrixError:
if !reflect.DeepEqual(tc.wantResponse, resp) {
@@ -533,7 +538,7 @@ func Test_register(t *testing.T) {
req = httptest.NewRequest(http.MethodPost, "/", body)
- resp = Register(req, userAPI, &base.Cfg.ClientAPI)
+ resp = Register(req, userAPI, &cfg.ClientAPI)
switch resp.JSON.(type) {
case *jsonerror.MatrixError:
@@ -576,17 +581,19 @@ func Test_register(t *testing.T) {
func TestRegisterUserWithDisplayName(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
- base.Cfg.Global.ServerName = "server"
-
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ defer close()
+ cfg.Global.ServerName = "server"
+
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
deviceName, deviceID := "deviceName", "deviceID"
expectedDisplayName := "DisplayName"
response := completeRegistration(
- base.Context(),
+ processCtx.Context(),
userAPI,
"user",
"server",
@@ -606,7 +613,7 @@ func TestRegisterUserWithDisplayName(t *testing.T) {
req := api.QueryProfileRequest{UserID: "@user:server"}
var res api.QueryProfileResponse
- err := userAPI.QueryProfile(base.Context(), &req, &res)
+ err := userAPI.QueryProfile(processCtx.Context(), &req, &res)
assert.NoError(t, err)
assert.Equal(t, expectedDisplayName, res.DisplayName)
})
@@ -614,14 +621,17 @@ func TestRegisterUserWithDisplayName(t *testing.T) {
func TestRegisterAdminUsingSharedSecret(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
- base.Cfg.Global.ServerName = "server"
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ defer close()
+ natsInstance := jetstream.NATSInstance{}
+ cfg.Global.ServerName = "server"
sharedSecret := "dendritetest"
- base.Cfg.ClientAPI.RegistrationSharedSecret = sharedSecret
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ cfg.ClientAPI.RegistrationSharedSecret = sharedSecret
+
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
expectedDisplayName := "rabbit"
jsonStr := []byte(`{"admin":true,"mac":"24dca3bba410e43fe64b9b5c28306693bf3baa9f","nonce":"759f047f312b99ff428b21d581256f8592b8976e58bc1b543972dc6147e529a79657605b52d7becd160ff5137f3de11975684319187e06901955f79e5a6c5a79","password":"wonderland","username":"alice","displayname":"rabbit"}`)
@@ -645,7 +655,7 @@ func TestRegisterAdminUsingSharedSecret(t *testing.T) {
ssrr := httptest.NewRequest(http.MethodPost, "/", body)
response := handleSharedSecretRegistration(
- &base.Cfg.ClientAPI,
+ &cfg.ClientAPI,
userAPI,
r,
ssrr,
@@ -654,7 +664,7 @@ func TestRegisterAdminUsingSharedSecret(t *testing.T) {
profilReq := api.QueryProfileRequest{UserID: "@alice:server"}
var profileRes api.QueryProfileResponse
- err = userAPI.QueryProfile(base.Context(), &profilReq, &profileRes)
+ err = userAPI.QueryProfile(processCtx.Context(), &profilReq, &profileRes)
assert.NoError(t, err)
assert.Equal(t, expectedDisplayName, profileRes.DisplayName)
})
diff --git a/clientapi/routing/routing.go b/clientapi/routing/routing.go
index 4ef2ac92..6a86980d 100644
--- a/clientapi/routing/routing.go
+++ b/clientapi/routing/routing.go
@@ -21,7 +21,6 @@ import (
"sync"
"github.com/gorilla/mux"
- "github.com/matrix-org/dendrite/setup/base"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@@ -50,8 +49,8 @@ import (
// applied:
// nolint: gocyclo
func Setup(
- base *base.BaseDendrite,
- cfg *config.ClientAPI,
+ routers httputil.Routers,
+ dendriteCfg *config.Dendrite,
rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
userAPI userapi.ClientUserAPI,
@@ -61,14 +60,16 @@ func Setup(
transactionsCache *transactions.Cache,
federationSender federationAPI.ClientFederationAPI,
extRoomsProvider api.ExtraPublicRoomsProvider,
- mscCfg *config.MSCs, natsClient *nats.Conn,
+ natsClient *nats.Conn, enableMetrics bool,
) {
- publicAPIMux := base.Routers.Client
- wkMux := base.Routers.WellKnown
- synapseAdminRouter := base.Routers.SynapseAdmin
- dendriteAdminRouter := base.Routers.DendriteAdmin
-
- if base.EnableMetrics {
+ cfg := &dendriteCfg.ClientAPI
+ mscCfg := &dendriteCfg.MSCs
+ publicAPIMux := routers.Client
+ wkMux := routers.WellKnown
+ synapseAdminRouter := routers.SynapseAdmin
+ dendriteAdminRouter := routers.DendriteAdmin
+
+ if enableMetrics {
prometheus.MustRegister(amtRegUsers, sendEventDuration)
}
@@ -656,7 +657,7 @@ func Setup(
).Methods(http.MethodGet, http.MethodPost, http.MethodOptions)
v3mux.Handle("/auth/{authType}/fallback/web",
- httputil.MakeHTMLAPI("auth_fallback", base.EnableMetrics, func(w http.ResponseWriter, req *http.Request) {
+ httputil.MakeHTMLAPI("auth_fallback", enableMetrics, func(w http.ResponseWriter, req *http.Request) {
vars := mux.Vars(req)
AuthFallback(w, req, vars["authType"], cfg)
}),
diff --git a/cmd/dendrite-demo-pinecone/conn/client.go b/cmd/dendrite-demo-pinecone/conn/client.go
index a91434f6..885de005 100644
--- a/cmd/dendrite-demo-pinecone/conn/client.go
+++ b/cmd/dendrite-demo-pinecone/conn/client.go
@@ -21,7 +21,7 @@ import (
"net/http"
"strings"
- "github.com/matrix-org/dendrite/setup/base"
+ "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
"nhooyr.io/websocket"
@@ -90,7 +90,7 @@ func createTransport(s *pineconeSessions.Sessions) *http.Transport {
}
func CreateClient(
- base *base.BaseDendrite, s *pineconeSessions.Sessions,
+ s *pineconeSessions.Sessions,
) *gomatrixserverlib.Client {
return gomatrixserverlib.NewClient(
gomatrixserverlib.WithTransport(createTransport(s)),
@@ -98,10 +98,10 @@ func CreateClient(
}
func CreateFederationClient(
- base *base.BaseDendrite, s *pineconeSessions.Sessions,
+ cfg *config.Dendrite, s *pineconeSessions.Sessions,
) *gomatrixserverlib.FederationClient {
return gomatrixserverlib.NewFederationClient(
- base.Cfg.Global.SigningIdentities(),
+ cfg.Global.SigningIdentities(),
gomatrixserverlib.WithTransport(createTransport(s)),
)
}
diff --git a/cmd/dendrite-demo-pinecone/main.go b/cmd/dendrite-demo-pinecone/main.go
index 7706a73f..7c710cbb 100644
--- a/cmd/dendrite-demo-pinecone/main.go
+++ b/cmd/dendrite-demo-pinecone/main.go
@@ -27,8 +27,11 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-pinecone/monolith"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/internal"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
@@ -87,9 +90,13 @@ func main() {
}
}
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+
enableMetrics := true
enableWebsockets := true
- p2pMonolith.SetupDendrite(cfg, *instancePort, *instanceRelayingEnabled, enableMetrics, enableWebsockets)
+ p2pMonolith.SetupDendrite(processCtx, cfg, cm, routers, *instancePort, *instanceRelayingEnabled, enableMetrics, enableWebsockets)
p2pMonolith.StartMonolith()
p2pMonolith.WaitForShutdown()
diff --git a/cmd/dendrite-demo-pinecone/monolith/monolith.go b/cmd/dendrite-demo-pinecone/monolith/monolith.go
index 5781d657..10a3493e 100644
--- a/cmd/dendrite-demo-pinecone/monolith/monolith.go
+++ b/cmd/dendrite-demo-pinecone/monolith/monolith.go
@@ -23,6 +23,7 @@ import (
"net"
"net/http"
"path/filepath"
+ "sync"
"time"
"github.com/gorilla/mux"
@@ -39,6 +40,7 @@ import (
"github.com/matrix-org/dendrite/federationapi/producers"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/relayapi"
relayAPI "github.com/matrix-org/dendrite/relayapi/api"
"github.com/matrix-org/dendrite/roomserver"
@@ -46,6 +48,7 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/userapi"
userAPI "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
@@ -61,13 +64,13 @@ import (
const SessionProtocol = "matrix"
type P2PMonolith struct {
- BaseDendrite *base.BaseDendrite
Sessions *pineconeSessions.Sessions
Multicast *pineconeMulticast.Multicast
ConnManager *pineconeConnections.ConnectionManager
Router *pineconeRouter.Router
EventChannel chan pineconeEvents.Event
RelayRetriever relay.RelayServerRetriever
+ ProcessCtx *process.ProcessContext
dendrite setup.Monolith
port int
@@ -77,6 +80,7 @@ type P2PMonolith struct {
listener net.Listener
httpListenAddr string
stopHandlingEvents chan bool
+ httpServerMu sync.Mutex
}
func GenerateDefaultConfig(sk ed25519.PrivateKey, storageDir string, cacheDir string, dbPrefix string) *config.Dendrite {
@@ -121,53 +125,52 @@ func (p *P2PMonolith) SetupPinecone(sk ed25519.PrivateKey) {
p.ConnManager = pineconeConnections.NewConnectionManager(p.Router, nil)
}
-func (p *P2PMonolith) SetupDendrite(cfg *config.Dendrite, port int, enableRelaying bool, enableMetrics bool, enableWebsockets bool) {
- if enableMetrics {
- p.BaseDendrite = base.NewBaseDendrite(cfg)
- } else {
- p.BaseDendrite = base.NewBaseDendrite(cfg, base.DisableMetrics)
- }
+func (p *P2PMonolith) SetupDendrite(
+ processCtx *process.ProcessContext, cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers,
+ port int, enableRelaying bool, enableMetrics bool, enableWebsockets bool) {
+
p.port = port
- p.BaseDendrite.ConfigureAdminEndpoints()
+ base.ConfigureAdminEndpoints(processCtx, routers)
- federation := conn.CreateFederationClient(p.BaseDendrite, p.Sessions)
+ federation := conn.CreateFederationClient(cfg, p.Sessions)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, enableMetrics)
- rsAPI := roomserver.NewInternalAPI(p.BaseDendrite, caches)
+ natsInstance := jetstream.NATSInstance{}
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, enableMetrics)
fsAPI := federationapi.NewInternalAPI(
- p.BaseDendrite, federation, rsAPI, caches, keyRing, true,
+ processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
- userAPI := userapi.NewInternalAPI(p.BaseDendrite, rsAPI, federation)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
- asAPI := appservice.NewInternalAPI(p.BaseDendrite, userAPI, rsAPI)
+ asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetFederationAPI(fsAPI, keyRing)
userProvider := users.NewPineconeUserProvider(p.Router, p.Sessions, userAPI, federation)
roomProvider := rooms.NewPineconeRoomProvider(p.Router, p.Sessions, fsAPI, federation)
- js, _ := p.BaseDendrite.NATS.Prepare(p.BaseDendrite.ProcessContext, &p.BaseDendrite.Cfg.Global.JetStream)
+ js, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
producer := &producers.SyncAPIProducer{
JetStream: js,
- TopicReceiptEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputReceiptEvent),
- TopicSendToDeviceEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
- TopicTypingEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputTypingEvent),
- TopicPresenceEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
- TopicDeviceListUpdate: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
- TopicSigningKeyUpdate: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
- Config: &p.BaseDendrite.Cfg.FederationAPI,
+ TopicReceiptEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputReceiptEvent),
+ TopicSendToDeviceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
+ TopicTypingEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputTypingEvent),
+ TopicPresenceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
+ TopicDeviceListUpdate: cfg.Global.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
+ TopicSigningKeyUpdate: cfg.Global.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
+ Config: &cfg.FederationAPI,
UserAPI: userAPI,
}
- relayAPI := relayapi.NewRelayInternalAPI(p.BaseDendrite, federation, rsAPI, keyRing, producer, enableRelaying, caches)
+ relayAPI := relayapi.NewRelayInternalAPI(cfg, cm, federation, rsAPI, keyRing, producer, enableRelaying, caches)
logrus.Infof("Relaying enabled: %v", relayAPI.RelayingEnabled())
p.dendrite = setup.Monolith{
- Config: p.BaseDendrite.Cfg,
- Client: conn.CreateClient(p.BaseDendrite, p.Sessions),
+ Config: cfg,
+ Client: conn.CreateClient(p.Sessions),
FedClient: federation,
KeyRing: keyRing,
@@ -179,9 +182,10 @@ func (p *P2PMonolith) SetupDendrite(cfg *config.Dendrite, port int, enableRelayi
ExtPublicRoomsProvider: roomProvider,
ExtUserDirectoryProvider: userProvider,
}
- p.dendrite.AddAllPublicRoutes(p.BaseDendrite, caches)
+ p.ProcessCtx = processCtx
+ p.dendrite.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, enableMetrics)
- p.setupHttpServers(userProvider, enableWebsockets)
+ p.setupHttpServers(userProvider, routers, enableWebsockets)
}
func (p *P2PMonolith) GetFederationAPI() federationAPI.FederationInternalAPI {
@@ -203,20 +207,22 @@ func (p *P2PMonolith) StartMonolith() {
func (p *P2PMonolith) Stop() {
logrus.Info("Stopping monolith")
- _ = p.BaseDendrite.Close()
+ p.ProcessCtx.ShutdownDendrite()
p.WaitForShutdown()
logrus.Info("Stopped monolith")
}
func (p *P2PMonolith) WaitForShutdown() {
- p.BaseDendrite.WaitForShutdown()
+ p.ProcessCtx.WaitForShutdown()
p.closeAllResources()
}
func (p *P2PMonolith) closeAllResources() {
logrus.Info("Closing monolith resources")
+ p.httpServerMu.Lock()
if p.httpServer != nil {
_ = p.httpServer.Shutdown(context.Background())
+ p.httpServerMu.Unlock()
}
select {
@@ -246,12 +252,12 @@ func (p *P2PMonolith) Addr() string {
return p.httpListenAddr
}
-func (p *P2PMonolith) setupHttpServers(userProvider *users.PineconeUserProvider, enableWebsockets bool) {
+func (p *P2PMonolith) setupHttpServers(userProvider *users.PineconeUserProvider, routers httputil.Routers, enableWebsockets bool) {
p.httpMux = mux.NewRouter().SkipClean(true).UseEncodedPath()
- p.httpMux.PathPrefix(httputil.PublicClientPathPrefix).Handler(p.BaseDendrite.Routers.Client)
- p.httpMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(p.BaseDendrite.Routers.Media)
- p.httpMux.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(p.BaseDendrite.Routers.DendriteAdmin)
- p.httpMux.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(p.BaseDendrite.Routers.SynapseAdmin)
+ p.httpMux.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
+ p.httpMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
+ p.httpMux.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
+ p.httpMux.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
if enableWebsockets {
wsUpgrader := websocket.Upgrader{
@@ -284,8 +290,8 @@ func (p *P2PMonolith) setupHttpServers(userProvider *users.PineconeUserProvider,
p.pineconeMux = mux.NewRouter().SkipClean(true).UseEncodedPath()
p.pineconeMux.PathPrefix(users.PublicURL).HandlerFunc(userProvider.FederatedUserProfiles)
- p.pineconeMux.PathPrefix(httputil.PublicFederationPathPrefix).Handler(p.BaseDendrite.Routers.Federation)
- p.pineconeMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(p.BaseDendrite.Routers.Media)
+ p.pineconeMux.PathPrefix(httputil.PublicFederationPathPrefix).Handler(routers.Federation)
+ p.pineconeMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
pHTTP := p.Sessions.Protocol(SessionProtocol).HTTP()
pHTTP.Mux().Handle(users.PublicURL, p.pineconeMux)
@@ -295,6 +301,7 @@ func (p *P2PMonolith) setupHttpServers(userProvider *users.PineconeUserProvider,
func (p *P2PMonolith) startHTTPServers() {
go func() {
+ p.httpServerMu.Lock()
// Build both ends of a HTTP multiplex.
p.httpServer = &http.Server{
Addr: ":0",
@@ -307,7 +314,7 @@ func (p *P2PMonolith) startHTTPServers() {
},
Handler: p.pineconeMux,
}
-
+ p.httpServerMu.Unlock()
pubkey := p.Router.PublicKey()
pubkeyString := hex.EncodeToString(pubkey[:])
logrus.Info("Listening on ", pubkeyString)
@@ -369,7 +376,7 @@ func (p *P2PMonolith) startEventHandler() {
ServerNames: []gomatrixserverlib.ServerName{gomatrixserverlib.ServerName(e.PeerID)},
}
res := &federationAPI.PerformWakeupServersResponse{}
- if err := p.dendrite.FederationAPI.PerformWakeupServers(p.BaseDendrite.Context(), req, res); err != nil {
+ if err := p.dendrite.FederationAPI.PerformWakeupServers(p.ProcessCtx.Context(), req, res); err != nil {
eLog.WithError(err).Error("Failed to wakeup destination", e.PeerID)
}
}
diff --git a/cmd/dendrite-demo-yggdrasil/main.go b/cmd/dendrite-demo-yggdrasil/main.go
index 62184719..9a195990 100644
--- a/cmd/dendrite-demo-yggdrasil/main.go
+++ b/cmd/dendrite-demo-yggdrasil/main.go
@@ -27,7 +27,11 @@ import (
"path/filepath"
"time"
+ "github.com/getsentry/sentry-go"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/gorilla/mux"
@@ -42,7 +46,7 @@ import (
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup"
- "github.com/matrix-org/dendrite/setup/base"
+ basepkg "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/mscs"
"github.com/matrix-org/dendrite/test"
@@ -58,6 +62,7 @@ var (
instanceDir = flag.String("dir", ".", "the directory to store the databases in (if --config not specified)")
)
+// nolint: gocyclo
func main() {
flag.Parse()
internal.SetupPprof()
@@ -143,36 +148,83 @@ func main() {
cfg.Global.ServerName = gomatrixserverlib.ServerName(hex.EncodeToString(pk))
cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID)
- base := base.NewBaseDendrite(cfg)
- base.ConfigureAdminEndpoints()
- defer base.Close() // nolint: errcheck
+ configErrors := &config.ConfigErrors{}
+ cfg.Verify(configErrors)
+ if len(*configErrors) > 0 {
+ for _, err := range *configErrors {
+ logrus.Errorf("Configuration error: %s", err)
+ }
+ logrus.Fatalf("Failed to start due to configuration errors")
+ }
+
+ internal.SetupStdLogging()
+ internal.SetupHookLogging(cfg.Logging)
+ internal.SetupPprof()
+
+ logrus.Infof("Dendrite version %s", internal.VersionString())
+
+ if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
+ logrus.Warn("Open registration is enabled")
+ }
+
+ closer, err := cfg.SetupTracing()
+ if err != nil {
+ logrus.WithError(err).Panicf("failed to start opentracing")
+ }
+ defer closer.Close() // nolint: errcheck
+
+ if cfg.Global.Sentry.Enabled {
+ logrus.Info("Setting up Sentry for debugging...")
+ err = sentry.Init(sentry.ClientOptions{
+ Dsn: cfg.Global.Sentry.DSN,
+ Environment: cfg.Global.Sentry.Environment,
+ Debug: true,
+ ServerName: string(cfg.Global.ServerName),
+ Release: "dendrite@" + internal.VersionString(),
+ AttachStacktrace: true,
+ })
+ if err != nil {
+ logrus.WithError(err).Panic("failed to start Sentry")
+ }
+ }
+
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+
+ basepkg.ConfigureAdminEndpoints(processCtx, routers)
+ defer func() {
+ processCtx.ShutdownDendrite()
+ processCtx.WaitForShutdown()
+ }() // nolint: errcheck
ygg, err := yggconn.Setup(sk, *instanceName, ".", *instancePeer, *instanceListen)
if err != nil {
panic(err)
}
- federation := ygg.CreateFederationClient(base)
+ federation := ygg.CreateFederationClient(cfg)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.EnableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
- userAPI := userapi.NewInternalAPI(base, rsAPI, federation)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
- asAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
+ asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
fsAPI := federationapi.NewInternalAPI(
- base, federation, rsAPI, caches, keyRing, true,
+ processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
rsAPI.SetFederationAPI(fsAPI, keyRing)
monolith := setup.Monolith{
- Config: base.Cfg,
- Client: ygg.CreateClient(base),
+ Config: cfg,
+ Client: ygg.CreateClient(),
FedClient: federation,
KeyRing: keyRing,
@@ -184,21 +236,21 @@ func main() {
ygg, fsAPI, federation,
),
}
- monolith.AddAllPublicRoutes(base, caches)
- if err := mscs.Enable(base, &monolith, caches); err != nil {
+ monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
+ if err := mscs.Enable(cfg, cm, routers, &monolith, caches); err != nil {
logrus.WithError(err).Fatalf("Failed to enable MSCs")
}
httpRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
- httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(base.Routers.Client)
- httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
- httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(base.Routers.DendriteAdmin)
- httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(base.Routers.SynapseAdmin)
+ httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
+ httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
+ httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
+ httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
embed.Embed(httpRouter, *instancePort, "Yggdrasil Demo")
yggRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
- yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(base.Routers.Federation)
- yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
+ yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(routers.Federation)
+ yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
// Build both ends of a HTTP multiplex.
httpServer := &http.Server{
@@ -232,5 +284,5 @@ func main() {
}()
// We want to block forever to let the HTTP and HTTPS handler serve the APIs
- base.WaitForShutdown()
+ basepkg.WaitForShutdown(processCtx)
}
diff --git a/cmd/dendrite-demo-yggdrasil/yggconn/client.go b/cmd/dendrite-demo-yggdrasil/yggconn/client.go
index 41a9ec12..51365547 100644
--- a/cmd/dendrite-demo-yggdrasil/yggconn/client.go
+++ b/cmd/dendrite-demo-yggdrasil/yggconn/client.go
@@ -4,7 +4,7 @@ import (
"net/http"
"time"
- "github.com/matrix-org/dendrite/setup/base"
+ "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
)
@@ -17,9 +17,7 @@ func (y *yggroundtripper) RoundTrip(req *http.Request) (*http.Response, error) {
return y.inner.RoundTrip(req)
}
-func (n *Node) CreateClient(
- base *base.BaseDendrite,
-) *gomatrixserverlib.Client {
+func (n *Node) CreateClient() *gomatrixserverlib.Client {
tr := &http.Transport{}
tr.RegisterProtocol(
"matrix", &yggroundtripper{
@@ -39,7 +37,7 @@ func (n *Node) CreateClient(
}
func (n *Node) CreateFederationClient(
- base *base.BaseDendrite,
+ cfg *config.Dendrite,
) *gomatrixserverlib.FederationClient {
tr := &http.Transport{}
tr.RegisterProtocol(
@@ -55,7 +53,7 @@ func (n *Node) CreateFederationClient(
},
)
return gomatrixserverlib.NewFederationClient(
- base.Cfg.Global.SigningIdentities(),
+ cfg.Global.SigningIdentities(),
gomatrixserverlib.WithTransport(tr),
)
}
diff --git a/cmd/dendrite-upgrade-tests/main.go b/cmd/dendrite-upgrade-tests/main.go
index 174a80a3..6a0e2179 100644
--- a/cmd/dendrite-upgrade-tests/main.go
+++ b/cmd/dendrite-upgrade-tests/main.go
@@ -259,10 +259,20 @@ func buildDendrite(httpClient *http.Client, dockerClient *client.Client, tmpDir
func getAndSortVersionsFromGithub(httpClient *http.Client) (semVers []*semver.Version, err error) {
u := "https://api.github.com/repos/matrix-org/dendrite/tags"
- res, err := httpClient.Get(u)
- if err != nil {
- return nil, err
+
+ var res *http.Response
+ for i := 0; i < 3; i++ {
+ res, err = httpClient.Get(u)
+ if err != nil {
+ return nil, err
+ }
+ if res.StatusCode == 200 {
+ break
+ }
+ log.Printf("Github API returned HTTP %d, retrying\n", res.StatusCode)
+ time.Sleep(time.Second * 5)
}
+
if res.StatusCode != 200 {
return nil, fmt.Errorf("%s returned HTTP %d", u, res.StatusCode)
}
diff --git a/cmd/dendrite/main.go b/cmd/dendrite/main.go
index 29290eb9..a22677eb 100644
--- a/cmd/dendrite/main.go
+++ b/cmd/dendrite/main.go
@@ -16,8 +16,16 @@ package main
import (
"flag"
+ "time"
+ "github.com/getsentry/sentry-go"
+ "github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
+ "github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/appservice"
@@ -66,25 +74,90 @@ func main() {
httpAddr = socket
}
- options := []basepkg.BaseDendriteOptions{}
+ configErrors := &config.ConfigErrors{}
+ cfg.Verify(configErrors)
+ if len(*configErrors) > 0 {
+ for _, err := range *configErrors {
+ logrus.Errorf("Configuration error: %s", err)
+ }
+ logrus.Fatalf("Failed to start due to configuration errors")
+ }
+ processCtx := process.NewProcessContext()
- base := basepkg.NewBaseDendrite(cfg, options...)
- defer base.Close() // nolint: errcheck
+ internal.SetupStdLogging()
+ internal.SetupHookLogging(cfg.Logging)
+ internal.SetupPprof()
- federation := base.CreateFederationClient()
+ basepkg.PlatformSanityChecks()
+
+ logrus.Infof("Dendrite version %s", internal.VersionString())
+ if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
+ logrus.Warn("Open registration is enabled")
+ }
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.EnableMetrics)
+ // create DNS cache
+ var dnsCache *gomatrixserverlib.DNSCache
+ if cfg.Global.DNSCache.Enabled {
+ dnsCache = gomatrixserverlib.NewDNSCache(
+ cfg.Global.DNSCache.CacheSize,
+ cfg.Global.DNSCache.CacheLifetime,
+ )
+ logrus.Infof(
+ "DNS cache enabled (size %d, lifetime %s)",
+ cfg.Global.DNSCache.CacheSize,
+ cfg.Global.DNSCache.CacheLifetime,
+ )
+ }
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ // setup tracing
+ closer, err := cfg.SetupTracing()
+ if err != nil {
+ logrus.WithError(err).Panicf("failed to start opentracing")
+ }
+ defer closer.Close() // nolint: errcheck
+
+ // setup sentry
+ if cfg.Global.Sentry.Enabled {
+ logrus.Info("Setting up Sentry for debugging...")
+ err = sentry.Init(sentry.ClientOptions{
+ Dsn: cfg.Global.Sentry.DSN,
+ Environment: cfg.Global.Sentry.Environment,
+ Debug: true,
+ ServerName: string(cfg.Global.ServerName),
+ Release: "dendrite@" + internal.VersionString(),
+ AttachStacktrace: true,
+ })
+ if err != nil {
+ logrus.WithError(err).Panic("failed to start Sentry")
+ }
+ go func() {
+ processCtx.ComponentStarted()
+ <-processCtx.WaitForShutdown()
+ if !sentry.Flush(time.Second * 5) {
+ logrus.Warnf("failed to flush all Sentry events!")
+ }
+ processCtx.ComponentFinished()
+ }()
+ }
+
+ federationClient := basepkg.CreateFederationClient(cfg, dnsCache)
+ httpClient := basepkg.CreateClient(cfg, dnsCache)
+
+ // prepare required dependencies
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+
+ caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
fsAPI := federationapi.NewInternalAPI(
- base, federation, rsAPI, caches, nil, false,
+ processCtx, cfg, cm, &natsInstance, federationClient, rsAPI, caches, nil, false,
)
keyRing := fsAPI.KeyRing()
- userAPI := userapi.NewInternalAPI(base, rsAPI, federation)
-
- asAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federationClient)
+ asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
// The underlying roomserver implementation needs to be able to call the fedsender.
// This is different to rsAPI which can be the http client which doesn't need this
@@ -94,9 +167,9 @@ func main() {
rsAPI.SetUserAPI(userAPI)
monolith := setup.Monolith{
- Config: base.Cfg,
- Client: base.CreateClient(),
- FedClient: federation,
+ Config: cfg,
+ Client: httpClient,
+ FedClient: federationClient,
KeyRing: keyRing,
AppserviceAPI: asAPI,
@@ -106,25 +179,25 @@ func main() {
RoomserverAPI: rsAPI,
UserAPI: userAPI,
}
- monolith.AddAllPublicRoutes(base, caches)
+ monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
- if len(base.Cfg.MSCs.MSCs) > 0 {
- if err := mscs.Enable(base, &monolith, caches); err != nil {
+ if len(cfg.MSCs.MSCs) > 0 {
+ if err := mscs.Enable(cfg, cm, routers, &monolith, caches); err != nil {
logrus.WithError(err).Fatalf("Failed to enable MSCs")
}
}
// Expose the matrix APIs directly rather than putting them under a /api path.
go func() {
- base.SetupAndServeHTTP(httpAddr, nil, nil)
+ basepkg.SetupAndServeHTTP(processCtx, cfg, routers, httpAddr, nil, nil)
}()
// Handle HTTPS if certificate and key are provided
if *unixSocket == "" && *certFile != "" && *keyFile != "" {
go func() {
- base.SetupAndServeHTTP(httpsAddr, certFile, keyFile)
+ basepkg.SetupAndServeHTTP(processCtx, cfg, routers, httpsAddr, certFile, keyFile)
}()
}
// We want to block forever to let the HTTP and HTTPS handler serve the APIs
- base.WaitForShutdown()
+ basepkg.WaitForShutdown(processCtx)
}
diff --git a/cmd/resolve-state/main.go b/cmd/resolve-state/main.go
index 099daaa9..09c0e690 100644
--- a/cmd/resolve-state/main.go
+++ b/cmd/resolve-state/main.go
@@ -10,12 +10,13 @@ import (
"time"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"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/setup"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
)
@@ -40,7 +41,7 @@ func main() {
Level: "error",
})
cfg.ClientAPI.RegistrationDisabled = true
- base := base.NewBaseDendrite(cfg, base.DisableMetrics)
+
args := flag.Args()
fmt.Println("Room version", *roomVersion)
@@ -54,8 +55,10 @@ func main() {
fmt.Println("Fetching", len(snapshotNIDs), "snapshot NIDs")
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
roomserverDB, err := storage.Open(
- base.ProcessContext.Context(), base.ConnectionManager, &cfg.RoomServer.Database,
+ processCtx.Context(), cm, &cfg.RoomServer.Database,
caching.NewRistrettoCache(128*1024*1024, time.Hour, true),
)
if err != nil {
diff --git a/federationapi/federationapi.go b/federationapi/federationapi.go
index 8a4237ba..3b5394a1 100644
--- a/federationapi/federationapi.go
+++ b/federationapi/federationapi.go
@@ -17,6 +17,10 @@ package federationapi
import (
"time"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/federationapi/api"
@@ -29,7 +33,6 @@ import (
"github.com/matrix-org/dendrite/federationapi/storage"
"github.com/matrix-org/dendrite/internal/caching"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
userapi "github.com/matrix-org/dendrite/userapi/api"
@@ -40,17 +43,21 @@ import (
// AddPublicRoutes sets up and registers HTTP handlers on the base API muxes for the FederationAPI component.
func AddPublicRoutes(
- base *base.BaseDendrite,
+ processContext *process.ProcessContext,
+ routers httputil.Routers,
+ dendriteConfig *config.Dendrite,
+ natsInstance *jetstream.NATSInstance,
userAPI userapi.FederationUserAPI,
federation *gomatrixserverlib.FederationClient,
keyRing gomatrixserverlib.JSONVerifier,
rsAPI roomserverAPI.FederationRoomserverAPI,
fedAPI federationAPI.FederationInternalAPI,
servers federationAPI.ServersInRoomProvider,
+ enableMetrics bool,
) {
- cfg := &base.Cfg.FederationAPI
- mscCfg := &base.Cfg.MSCs
- js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
+ cfg := &dendriteConfig.FederationAPI
+ mscCfg := &dendriteConfig.MSCs
+ js, _ := natsInstance.Prepare(processContext, &cfg.Matrix.JetStream)
producer := &producers.SyncAPIProducer{
JetStream: js,
TopicReceiptEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
@@ -75,26 +82,30 @@ func AddPublicRoutes(
}
routing.Setup(
- base,
+ routers,
+ dendriteConfig,
rsAPI, f, keyRing,
federation, userAPI, mscCfg,
- servers, producer,
+ servers, producer, enableMetrics,
)
}
// NewInternalAPI returns a concerete implementation of the internal API. Callers
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
func NewInternalAPI(
- base *base.BaseDendrite,
+ processContext *process.ProcessContext,
+ dendriteCfg *config.Dendrite,
+ cm sqlutil.Connections,
+ natsInstance *jetstream.NATSInstance,
federation api.FederationClient,
rsAPI roomserverAPI.FederationRoomserverAPI,
caches *caching.Caches,
keyRing *gomatrixserverlib.KeyRing,
resetBlacklist bool,
) api.FederationInternalAPI {
- cfg := &base.Cfg.FederationAPI
+ cfg := &dendriteCfg.FederationAPI
- federationDB, err := storage.NewDatabase(base.ProcessContext.Context(), base.ConnectionManager, &cfg.Database, caches, base.Cfg.Global.IsLocalServerName)
+ federationDB, err := storage.NewDatabase(processContext.Context(), cm, &cfg.Database, caches, dendriteCfg.Global.IsLocalServerName)
if err != nil {
logrus.WithError(err).Panic("failed to connect to federation sender db")
}
@@ -108,51 +119,51 @@ func NewInternalAPI(
cfg.FederationMaxRetries+1,
cfg.P2PFederationRetriesUntilAssumedOffline+1)
- js, nats := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
+ js, nats := natsInstance.Prepare(processContext, &cfg.Matrix.JetStream)
- signingInfo := base.Cfg.Global.SigningIdentities()
+ signingInfo := dendriteCfg.Global.SigningIdentities()
queues := queue.NewOutgoingQueues(
- federationDB, base.ProcessContext,
+ federationDB, processContext,
cfg.Matrix.DisableFederation,
cfg.Matrix.ServerName, federation, rsAPI, &stats,
signingInfo,
)
rsConsumer := consumers.NewOutputRoomEventConsumer(
- base.ProcessContext, cfg, js, nats, queues,
+ processContext, cfg, js, nats, queues,
federationDB, rsAPI,
)
if err = rsConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start room server consumer")
}
tsConsumer := consumers.NewOutputSendToDeviceConsumer(
- base.ProcessContext, cfg, js, queues, federationDB,
+ processContext, cfg, js, queues, federationDB,
)
if err = tsConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start send-to-device consumer")
}
receiptConsumer := consumers.NewOutputReceiptConsumer(
- base.ProcessContext, cfg, js, queues, federationDB,
+ processContext, cfg, js, queues, federationDB,
)
if err = receiptConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start receipt consumer")
}
typingConsumer := consumers.NewOutputTypingConsumer(
- base.ProcessContext, cfg, js, queues, federationDB,
+ processContext, cfg, js, queues, federationDB,
)
if err = typingConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start typing consumer")
}
keyConsumer := consumers.NewKeyChangeConsumer(
- base.ProcessContext, &base.Cfg.KeyServer, js, queues, federationDB, rsAPI,
+ processContext, &dendriteCfg.KeyServer, js, queues, federationDB, rsAPI,
)
if err = keyConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start key server consumer")
}
presenceConsumer := consumers.NewOutputPresenceConsumer(
- base.ProcessContext, cfg, js, queues, federationDB, rsAPI,
+ processContext, cfg, js, queues, federationDB, rsAPI,
)
if err = presenceConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start presence consumer")
@@ -161,7 +172,7 @@ func NewInternalAPI(
var cleanExpiredEDUs func()
cleanExpiredEDUs = func() {
logrus.Infof("Cleaning expired EDUs")
- if err := federationDB.DeleteExpiredEDUs(base.Context()); err != nil {
+ if err := federationDB.DeleteExpiredEDUs(processContext.Context()); err != nil {
logrus.WithError(err).Error("Failed to clean expired EDUs")
}
time.AfterFunc(time.Hour, cleanExpiredEDUs)
diff --git a/federationapi/federationapi_keys_test.go b/federationapi/federationapi_keys_test.go
index bb6ee893..425cdbb9 100644
--- a/federationapi/federationapi_keys_test.go
+++ b/federationapi/federationapi_keys_test.go
@@ -12,12 +12,14 @@ import (
"testing"
"time"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/routing"
"github.com/matrix-org/dendrite/internal/caching"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
)
@@ -65,7 +67,7 @@ func TestMain(m *testing.M) {
// Create a new cache but don't enable prometheus!
s.cache = caching.NewRistrettoCache(8*1024*1024, time.Hour, false)
-
+ natsInstance := jetstream.NATSInstance{}
// Create a temporary directory for JetStream.
d, err := os.MkdirTemp("./", "jetstream*")
if err != nil {
@@ -109,8 +111,9 @@ func TestMain(m *testing.M) {
)
// Finally, build the server key APIs.
- sbase := base.NewBaseDendrite(cfg, base.DisableMetrics)
- s.api = NewInternalAPI(sbase, s.fedclient, nil, s.cache, nil, true)
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ s.api = NewInternalAPI(processCtx, cfg, cm, &natsInstance, s.fedclient, nil, s.cache, nil, true)
}
// Now that we have built our server key APIs, start the
diff --git a/federationapi/federationapi_test.go b/federationapi/federationapi_test.go
index 8aea96a7..d39f4051 100644
--- a/federationapi/federationapi_test.go
+++ b/federationapi/federationapi_test.go
@@ -11,6 +11,8 @@ import (
"time"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
@@ -19,8 +21,6 @@ import (
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/internal"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
- "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
@@ -163,22 +163,24 @@ func TestFederationAPIJoinThenKeyUpdate(t *testing.T) {
}
func testFederationAPIJoinThenKeyUpdate(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- base.Cfg.FederationAPI.PreferDirectFetch = true
- base.Cfg.FederationAPI.KeyPerspectives = nil
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ cfg.FederationAPI.PreferDirectFetch = true
+ cfg.FederationAPI.KeyPerspectives = nil
defer close()
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
serverA := gomatrixserverlib.ServerName("server.a")
serverAKeyID := gomatrixserverlib.KeyID("ed25519:servera")
serverAPrivKey := test.PrivateKeyA
creator := test.NewUser(t, test.WithSigningServer(serverA, serverAKeyID, serverAPrivKey))
- myServer := base.Cfg.Global.ServerName
- myServerKeyID := base.Cfg.Global.KeyID
- myServerPrivKey := base.Cfg.Global.PrivateKey
+ myServer := cfg.Global.ServerName
+ myServerKeyID := cfg.Global.KeyID
+ myServerPrivKey := cfg.Global.PrivateKey
joiningUser := test.NewUser(t, test.WithSigningServer(myServer, myServerKeyID, myServerPrivKey))
fmt.Printf("creator: %v joining user: %v\n", creator.ID, joiningUser.ID)
room := test.NewRoom(t, creator)
@@ -214,7 +216,7 @@ func testFederationAPIJoinThenKeyUpdate(t *testing.T, dbType test.DBType) {
},
},
}
- fsapi := federationapi.NewInternalAPI(base, fc, rsapi, caches, nil, false)
+ fsapi := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, fc, rsapi, caches, nil, false)
var resp api.PerformJoinResponse
fsapi.PerformJoin(context.Background(), &api.PerformJoinRequest{
@@ -247,7 +249,7 @@ func testFederationAPIJoinThenKeyUpdate(t *testing.T, dbType test.DBType) {
}
msg := &nats.Msg{
- Subject: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
+ Subject: cfg.Global.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
Header: nats.Header{},
Data: b,
}
@@ -265,30 +267,6 @@ func testFederationAPIJoinThenKeyUpdate(t *testing.T, dbType test.DBType) {
// Tests that event IDs with '/' in them (escaped as %2F) are correctly passed to the right handler and don't 404.
// Relevant for v3 rooms and a cause of flakey sytests as the IDs are randomly generated.
func TestRoomsV3URLEscapeDoNot404(t *testing.T) {
- _, privKey, _ := ed25519.GenerateKey(nil)
- cfg := &config.Dendrite{}
- cfg.Defaults(config.DefaultOpts{
- Generate: true,
- SingleDatabase: false,
- })
- cfg.Global.KeyID = gomatrixserverlib.KeyID("ed25519:auto")
- cfg.Global.ServerName = gomatrixserverlib.ServerName("localhost")
- cfg.Global.PrivateKey = privKey
- cfg.Global.JetStream.InMemory = true
- b := base.NewBaseDendrite(cfg, base.DisableMetrics)
- keyRing := &test.NopJSONVerifier{}
- // TODO: This is pretty fragile, as if anything calls anything on these nils this test will break.
- // Unfortunately, it makes little sense to instantiate these dependencies when we just want to test routing.
- federationapi.AddPublicRoutes(b, nil, nil, keyRing, nil, &internal.FederationInternalAPI{}, nil)
- baseURL, cancel := test.ListenAndServe(t, b.Routers.Federation, true)
- defer cancel()
- serverName := gomatrixserverlib.ServerName(strings.TrimPrefix(baseURL, "https://"))
-
- fedCli := gomatrixserverlib.NewFederationClient(
- cfg.Global.SigningIdentities(),
- gomatrixserverlib.WithSkipVerify(true),
- )
-
testCases := []struct {
roomVer gomatrixserverlib.RoomVersion
eventJSON string
@@ -317,6 +295,29 @@ func TestRoomsV3URLEscapeDoNot404(t *testing.T) {
},
}
+ cfg, processCtx, close := testrig.CreateConfig(t, test.DBTypeSQLite)
+ defer close()
+ routers := httputil.NewRouters()
+
+ _, privKey, _ := ed25519.GenerateKey(nil)
+ cfg.Global.KeyID = gomatrixserverlib.KeyID("ed25519:auto")
+ cfg.Global.ServerName = gomatrixserverlib.ServerName("localhost")
+ cfg.Global.PrivateKey = privKey
+ cfg.Global.JetStream.InMemory = true
+ keyRing := &test.NopJSONVerifier{}
+ natsInstance := jetstream.NATSInstance{}
+ // TODO: This is pretty fragile, as if anything calls anything on these nils this test will break.
+ // Unfortunately, it makes little sense to instantiate these dependencies when we just want to test routing.
+ federationapi.AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, nil, keyRing, nil, &internal.FederationInternalAPI{}, nil, caching.DisableMetrics)
+ baseURL, cancel := test.ListenAndServe(t, routers.Federation, true)
+ defer cancel()
+ serverName := gomatrixserverlib.ServerName(strings.TrimPrefix(baseURL, "https://"))
+
+ fedCli := gomatrixserverlib.NewFederationClient(
+ cfg.Global.SigningIdentities(),
+ gomatrixserverlib.WithSkipVerify(true),
+ )
+
for _, tc := range testCases {
ev, err := gomatrixserverlib.NewEventFromTrustedJSON([]byte(tc.eventJSON), false, tc.roomVer)
if err != nil {
diff --git a/federationapi/queue/queue_test.go b/federationapi/queue/queue_test.go
index c3fb1056..55d1df4a 100644
--- a/federationapi/queue/queue_test.go
+++ b/federationapi/queue/queue_test.go
@@ -22,6 +22,7 @@ import (
"time"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/test/testrig"
"go.uber.org/atomic"
"gotest.tools/v3/poll"
@@ -41,18 +42,19 @@ import (
func mustCreateFederationDatabase(t *testing.T, dbType test.DBType, realDatabase bool) (storage.Database, *process.ProcessContext, func()) {
if realDatabase {
// Real Database/s
- b, baseClose := testrig.CreateBaseDendrite(t, dbType)
- caches := caching.NewRistrettoCache(b.Cfg.Global.Cache.EstimatedMaxSize, b.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
connStr, dbClose := test.PrepareDBConnectionString(t, dbType)
- db, err := storage.NewDatabase(b.ProcessContext.Context(), b.ConnectionManager, &config.DatabaseOptions{
+ db, err := storage.NewDatabase(processCtx.Context(), cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
- }, caches, b.Cfg.Global.IsLocalServerName)
+ }, caches, cfg.Global.IsLocalServerName)
if err != nil {
t.Fatalf("NewDatabase returned %s", err)
}
- return db, b.ProcessContext, func() {
+ return db, processCtx, func() {
+ close()
dbClose()
- baseClose()
}
} else {
// Fake Database
diff --git a/federationapi/routing/profile_test.go b/federationapi/routing/profile_test.go
index df494a74..d5e9997f 100644
--- a/federationapi/routing/profile_test.go
+++ b/federationapi/routing/profile_test.go
@@ -27,7 +27,10 @@ import (
fedAPI "github.com/matrix-org/dendrite/federationapi"
fedInternal "github.com/matrix-org/dendrite/federationapi/internal"
"github.com/matrix-org/dendrite/federationapi/routing"
+ "github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
userAPI "github.com/matrix-org/dendrite/userapi/api"
@@ -46,23 +49,26 @@ func (u *fakeUserAPI) QueryProfile(ctx context.Context, req *userAPI.QueryProfil
func TestHandleQueryProfile(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
defer close()
fedMux := mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicFederationPathPrefix).Subrouter().UseEncodedPath()
- base.Routers.Federation = fedMux
- base.Cfg.FederationAPI.Matrix.SigningIdentity.ServerName = testOrigin
- base.Cfg.FederationAPI.Matrix.Metrics.Enabled = false
+ natsInstance := jetstream.NATSInstance{}
+ routers.Federation = fedMux
+ cfg.FederationAPI.Matrix.SigningIdentity.ServerName = testOrigin
+ cfg.FederationAPI.Matrix.Metrics.Enabled = false
fedClient := fakeFedClient{}
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
- fedapi := fedAPI.NewInternalAPI(base, &fedClient, nil, nil, keyRing, true)
+ fedapi := fedAPI.NewInternalAPI(processCtx, cfg, cm, &natsInstance, &fedClient, nil, nil, keyRing, true)
userapi := fakeUserAPI{}
r, ok := fedapi.(*fedInternal.FederationInternalAPI)
if !ok {
panic("This is a programming error.")
}
- routing.Setup(base, nil, r, keyRing, &fedClient, &userapi, &base.Cfg.MSCs, nil, nil)
+ routing.Setup(routers, cfg, nil, r, keyRing, &fedClient, &userapi, &cfg.MSCs, nil, nil, caching.DisableMetrics)
handler := fedMux.Get(routing.QueryProfileRouteName).GetHandler().ServeHTTP
_, sk, _ := ed25519.GenerateKey(nil)
diff --git a/federationapi/routing/query_test.go b/federationapi/routing/query_test.go
index 69cf7047..6e702963 100644
--- a/federationapi/routing/query_test.go
+++ b/federationapi/routing/query_test.go
@@ -28,7 +28,10 @@ import (
fedclient "github.com/matrix-org/dendrite/federationapi/api"
fedInternal "github.com/matrix-org/dendrite/federationapi/internal"
"github.com/matrix-org/dendrite/federationapi/routing"
+ "github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/gomatrixserverlib"
@@ -46,23 +49,26 @@ func (f *fakeFedClient) LookupRoomAlias(ctx context.Context, origin, s gomatrixs
func TestHandleQueryDirectory(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
defer close()
fedMux := mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicFederationPathPrefix).Subrouter().UseEncodedPath()
- base.Routers.Federation = fedMux
- base.Cfg.FederationAPI.Matrix.SigningIdentity.ServerName = testOrigin
- base.Cfg.FederationAPI.Matrix.Metrics.Enabled = false
+ natsInstance := jetstream.NATSInstance{}
+ routers.Federation = fedMux
+ cfg.FederationAPI.Matrix.SigningIdentity.ServerName = testOrigin
+ cfg.FederationAPI.Matrix.Metrics.Enabled = false
fedClient := fakeFedClient{}
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
- fedapi := fedAPI.NewInternalAPI(base, &fedClient, nil, nil, keyRing, true)
+ fedapi := fedAPI.NewInternalAPI(processCtx, cfg, cm, &natsInstance, &fedClient, nil, nil, keyRing, true)
userapi := fakeUserAPI{}
r, ok := fedapi.(*fedInternal.FederationInternalAPI)
if !ok {
panic("This is a programming error.")
}
- routing.Setup(base, nil, r, keyRing, &fedClient, &userapi, &base.Cfg.MSCs, nil, nil)
+ routing.Setup(routers, cfg, nil, r, keyRing, &fedClient, &userapi, &cfg.MSCs, nil, nil, caching.DisableMetrics)
handler := fedMux.Get(routing.QueryDirectoryRouteName).GetHandler().ServeHTTP
_, sk, _ := ed25519.GenerateKey(nil)
diff --git a/federationapi/routing/routing.go b/federationapi/routing/routing.go
index c86d18d2..a1f943e7 100644
--- a/federationapi/routing/routing.go
+++ b/federationapi/routing/routing.go
@@ -31,7 +31,6 @@ import (
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/roomserver/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
@@ -55,7 +54,8 @@ const (
// applied:
// nolint: gocyclo
func Setup(
- base *base.BaseDendrite,
+ routers httputil.Routers,
+ dendriteCfg *config.Dendrite,
rsAPI roomserverAPI.FederationRoomserverAPI,
fsAPI *fedInternal.FederationInternalAPI,
keys gomatrixserverlib.JSONVerifier,
@@ -63,14 +63,14 @@ func Setup(
userAPI userapi.FederationUserAPI,
mscCfg *config.MSCs,
servers federationAPI.ServersInRoomProvider,
- producer *producers.SyncAPIProducer,
+ producer *producers.SyncAPIProducer, enableMetrics bool,
) {
- fedMux := base.Routers.Federation
- keyMux := base.Routers.Keys
- wkMux := base.Routers.WellKnown
- cfg := &base.Cfg.FederationAPI
+ fedMux := routers.Federation
+ keyMux := routers.Keys
+ wkMux := routers.WellKnown
+ cfg := &dendriteCfg.FederationAPI
- if base.EnableMetrics {
+ if enableMetrics {
prometheus.MustRegister(
internal.PDUCountTotal, internal.EDUCountTotal,
)
diff --git a/federationapi/routing/send_test.go b/federationapi/routing/send_test.go
index 53e1399a..28fa6d6d 100644
--- a/federationapi/routing/send_test.go
+++ b/federationapi/routing/send_test.go
@@ -25,7 +25,10 @@ import (
fedAPI "github.com/matrix-org/dendrite/federationapi"
fedInternal "github.com/matrix-org/dendrite/federationapi/internal"
"github.com/matrix-org/dendrite/federationapi/routing"
+ "github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/gomatrixserverlib"
@@ -44,21 +47,24 @@ type sendContent struct {
func TestHandleSend(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
defer close()
fedMux := mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicFederationPathPrefix).Subrouter().UseEncodedPath()
- base.Routers.Federation = fedMux
- base.Cfg.FederationAPI.Matrix.SigningIdentity.ServerName = testOrigin
- base.Cfg.FederationAPI.Matrix.Metrics.Enabled = false
- fedapi := fedAPI.NewInternalAPI(base, nil, nil, nil, nil, true)
+ natsInstance := jetstream.NATSInstance{}
+ routers.Federation = fedMux
+ cfg.FederationAPI.Matrix.SigningIdentity.ServerName = testOrigin
+ cfg.FederationAPI.Matrix.Metrics.Enabled = false
+ fedapi := fedAPI.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, nil, nil, true)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
r, ok := fedapi.(*fedInternal.FederationInternalAPI)
if !ok {
panic("This is a programming error.")
}
- routing.Setup(base, nil, r, keyRing, nil, nil, &base.Cfg.MSCs, nil, nil)
+ routing.Setup(routers, cfg, nil, r, keyRing, nil, nil, &cfg.MSCs, nil, nil, caching.DisableMetrics)
handler := fedMux.Get(routing.SendRouteName).GetHandler().ServeHTTP
_, sk, _ := ed25519.GenerateKey(nil)
diff --git a/federationapi/storage/storage_test.go b/federationapi/storage/storage_test.go
index a02b71b6..74863c07 100644
--- a/federationapi/storage/storage_test.go
+++ b/federationapi/storage/storage_test.go
@@ -20,7 +20,7 @@ func mustCreateFederationDatabase(t *testing.T, dbType test.DBType) (storage.Dat
caches := caching.NewRistrettoCache(8*1024*1024, time.Hour, false)
connStr, dbClose := test.PrepareDBConnectionString(t, dbType)
ctx := context.Background()
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewDatabase(ctx, cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, caches, func(server gomatrixserverlib.ServerName) bool { return server == "localhost" })
diff --git a/internal/sqlutil/connection_manager.go b/internal/sqlutil/connection_manager.go
index cefd9f80..934a2954 100644
--- a/internal/sqlutil/connection_manager.go
+++ b/internal/sqlutil/connection_manager.go
@@ -19,15 +19,21 @@ import (
"fmt"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
)
type Connections struct {
- db *sql.DB
- writer Writer
+ db *sql.DB
+ writer Writer
+ globalConfig config.DatabaseOptions
+ processContext *process.ProcessContext
}
-func NewConnectionManager() Connections {
- return Connections{}
+func NewConnectionManager(processCtx *process.ProcessContext, globalConfig config.DatabaseOptions) Connections {
+ return Connections{
+ globalConfig: globalConfig,
+ processContext: processCtx,
+ }
}
func (c *Connections) Connection(dbProperties *config.DatabaseOptions) (*sql.DB, Writer, error) {
@@ -35,14 +41,29 @@ func (c *Connections) Connection(dbProperties *config.DatabaseOptions) (*sql.DB,
if dbProperties.ConnectionString.IsSQLite() {
writer = NewExclusiveWriter()
}
+ var err error
+ if dbProperties.ConnectionString == "" {
+ // if no connectionString was provided, try the global one
+ dbProperties = &c.globalConfig
+ }
if dbProperties.ConnectionString != "" || c.db == nil {
- var err error
// Open a new database connection using the supplied config.
c.db, err = Open(dbProperties, writer)
if err != nil {
return nil, nil, err
}
c.writer = writer
+ go func() {
+ if c.processContext == nil {
+ return
+ }
+ // If we have a ProcessContext, start a component and wait for
+ // Dendrite to shut down to cleanly close the database connection.
+ c.processContext.ComponentStarted()
+ <-c.processContext.WaitForShutdown()
+ _ = c.db.Close()
+ c.processContext.ComponentFinished()
+ }()
return c.db, c.writer, nil
}
if c.db != nil && c.writer != nil {
diff --git a/internal/sqlutil/connection_manager_test.go b/internal/sqlutil/connection_manager_test.go
index 610629d5..a9ac8d57 100644
--- a/internal/sqlutil/connection_manager_test.go
+++ b/internal/sqlutil/connection_manager_test.go
@@ -13,9 +13,9 @@ func TestConnectionManager(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
conStr, close := test.PrepareDBConnectionString(t, dbType)
t.Cleanup(close)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
- dbProps := &config.DatabaseOptions{ConnectionString: config.DataSource(string(conStr))}
+ dbProps := &config.DatabaseOptions{ConnectionString: config.DataSource(conStr)}
db, writer, err := cm.Connection(dbProps)
if err != nil {
t.Fatal(err)
@@ -47,8 +47,8 @@ func TestConnectionManager(t *testing.T) {
}
// test invalid connection string configured
- cm = sqlutil.NewConnectionManager()
- _, _, err = cm.Connection(&config.DatabaseOptions{ConnectionString: "http://"})
+ cm2 := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
+ _, _, err = cm2.Connection(&config.DatabaseOptions{ConnectionString: "http://"})
if err == nil {
t.Fatal("expected an error but got none")
}
diff --git a/mediaapi/mediaapi.go b/mediaapi/mediaapi.go
index 42e0ea88..5d517ef2 100644
--- a/mediaapi/mediaapi.go
+++ b/mediaapi/mediaapi.go
@@ -15,9 +15,11 @@
package mediaapi
import (
+ "github.com/gorilla/mux"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/mediaapi/routing"
"github.com/matrix-org/dendrite/mediaapi/storage"
- "github.com/matrix-org/dendrite/setup/base"
+ "github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
@@ -25,19 +27,18 @@ import (
// AddPublicRoutes sets up and registers HTTP handlers for the MediaAPI component.
func AddPublicRoutes(
- base *base.BaseDendrite,
+ mediaRouter *mux.Router,
+ cm sqlutil.Connections,
+ cfg *config.Dendrite,
userAPI userapi.MediaUserAPI,
client *gomatrixserverlib.Client,
) {
- cfg := &base.Cfg.MediaAPI
- rateCfg := &base.Cfg.ClientAPI.RateLimiting
-
- mediaDB, err := storage.NewMediaAPIDatasource(base.ConnectionManager, &cfg.Database)
+ mediaDB, err := storage.NewMediaAPIDatasource(cm, &cfg.MediaAPI.Database)
if err != nil {
logrus.WithError(err).Panicf("failed to connect to media db")
}
routing.Setup(
- base.Routers.Media, cfg, rateCfg, mediaDB, userAPI, client,
+ mediaRouter, cfg, mediaDB, userAPI, client,
)
}
diff --git a/mediaapi/routing/routing.go b/mediaapi/routing/routing.go
index 8a4a5afc..98e6a82e 100644
--- a/mediaapi/routing/routing.go
+++ b/mediaapi/routing/routing.go
@@ -45,13 +45,12 @@ type configResponse struct {
// nolint: gocyclo
func Setup(
publicAPIMux *mux.Router,
- cfg *config.MediaAPI,
- rateLimit *config.RateLimiting,
+ cfg *config.Dendrite,
db storage.Database,
userAPI userapi.MediaUserAPI,
client *gomatrixserverlib.Client,
) {
- rateLimits := httputil.NewRateLimits(rateLimit)
+ rateLimits := httputil.NewRateLimits(&cfg.ClientAPI.RateLimiting)
v3mux := publicAPIMux.PathPrefix("/{apiversion:(?:r0|v1|v3)}/").Subrouter()
@@ -65,7 +64,7 @@ func Setup(
if r := rateLimits.Limit(req, dev); r != nil {
return *r
}
- return Upload(req, cfg, dev, db, activeThumbnailGeneration)
+ return Upload(req, &cfg.MediaAPI, dev, db, activeThumbnailGeneration)
},
)
@@ -73,8 +72,8 @@ func Setup(
if r := rateLimits.Limit(req, device); r != nil {
return *r
}
- respondSize := &cfg.MaxFileSizeBytes
- if cfg.MaxFileSizeBytes == 0 {
+ respondSize := &cfg.MediaAPI.MaxFileSizeBytes
+ if cfg.MediaAPI.MaxFileSizeBytes == 0 {
respondSize = nil
}
return util.JSONResponse{
@@ -90,12 +89,12 @@ func Setup(
MXCToResult: map[string]*types.RemoteRequestResult{},
}
- downloadHandler := makeDownloadAPI("download", cfg, rateLimits, db, client, activeRemoteRequests, activeThumbnailGeneration)
+ downloadHandler := makeDownloadAPI("download", &cfg.MediaAPI, rateLimits, db, client, activeRemoteRequests, activeThumbnailGeneration)
v3mux.Handle("/download/{serverName}/{mediaId}", downloadHandler).Methods(http.MethodGet, http.MethodOptions)
v3mux.Handle("/download/{serverName}/{mediaId}/{downloadName}", downloadHandler).Methods(http.MethodGet, http.MethodOptions)
v3mux.Handle("/thumbnail/{serverName}/{mediaId}",
- makeDownloadAPI("thumbnail", cfg, rateLimits, db, client, activeRemoteRequests, activeThumbnailGeneration),
+ makeDownloadAPI("thumbnail", &cfg.MediaAPI, rateLimits, db, client, activeRemoteRequests, activeThumbnailGeneration),
).Methods(http.MethodGet, http.MethodOptions)
}
diff --git a/mediaapi/routing/upload_test.go b/mediaapi/routing/upload_test.go
index d4fb45d1..d088950c 100644
--- a/mediaapi/routing/upload_test.go
+++ b/mediaapi/routing/upload_test.go
@@ -50,7 +50,7 @@ func Test_uploadRequest_doUpload(t *testing.T) {
// create testdata folder and remove when done
_ = os.Mkdir(testdataPath, os.ModePerm)
defer fileutils.RemoveDir(types.Path(testdataPath), nil)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewMediaAPIDatasource(cm, &config.DatabaseOptions{
ConnectionString: "file::memory:?cache=shared",
MaxOpenConnections: 100,
diff --git a/mediaapi/storage/storage_test.go b/mediaapi/storage/storage_test.go
index 11febd27..8cd29a54 100644
--- a/mediaapi/storage/storage_test.go
+++ b/mediaapi/storage/storage_test.go
@@ -14,7 +14,7 @@ import (
func mustCreateDatabase(t *testing.T, dbType test.DBType) (storage.Database, func()) {
connStr, close := test.PrepareDBConnectionString(t, dbType)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewMediaAPIDatasource(cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
})
diff --git a/relayapi/relayapi.go b/relayapi/relayapi.go
index 925fc031..fdc366e4 100644
--- a/relayapi/relayapi.go
+++ b/relayapi/relayapi.go
@@ -17,24 +17,25 @@ package relayapi
import (
"github.com/matrix-org/dendrite/federationapi/producers"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/relayapi/api"
"github.com/matrix-org/dendrite/relayapi/internal"
"github.com/matrix-org/dendrite/relayapi/routing"
"github.com/matrix-org/dendrite/relayapi/storage"
rsAPI "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
+ "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
)
// AddPublicRoutes sets up and registers HTTP handlers on the base API muxes for the FederationAPI component.
func AddPublicRoutes(
- base *base.BaseDendrite,
+ routers httputil.Routers,
+ dendriteCfg *config.Dendrite,
keyRing gomatrixserverlib.JSONVerifier,
relayAPI api.RelayInternalAPI,
) {
- fedCfg := &base.Cfg.FederationAPI
-
relay, ok := relayAPI.(*internal.RelayInternalAPI)
if !ok {
panic("relayapi.AddPublicRoutes called with a RelayInternalAPI impl which was not " +
@@ -42,15 +43,16 @@ func AddPublicRoutes(
}
routing.Setup(
- base.Routers.Federation,
- fedCfg,
+ routers.Federation,
+ &dendriteCfg.FederationAPI,
relay,
keyRing,
)
}
func NewRelayInternalAPI(
- base *base.BaseDendrite,
+ dendriteCfg *config.Dendrite,
+ cm sqlutil.Connections,
fedClient *gomatrixserverlib.FederationClient,
rsAPI rsAPI.RoomserverInternalAPI,
keyRing *gomatrixserverlib.KeyRing,
@@ -58,8 +60,7 @@ func NewRelayInternalAPI(
relayingEnabled bool,
caches caching.FederationCache,
) api.RelayInternalAPI {
- cfg := &base.Cfg.RelayAPI
- relayDB, err := storage.NewDatabase(base.ConnectionManager, &cfg.Database, caches, base.Cfg.Global.IsLocalServerName)
+ relayDB, err := storage.NewDatabase(cm, &dendriteCfg.RelayAPI.Database, caches, dendriteCfg.Global.IsLocalServerName)
if err != nil {
logrus.WithError(err).Panic("failed to connect to relay db")
}
@@ -70,8 +71,8 @@ func NewRelayInternalAPI(
rsAPI,
keyRing,
producer,
- base.Cfg.Global.Presence.EnableInbound,
- base.Cfg.Global.ServerName,
+ dendriteCfg.Global.Presence.EnableInbound,
+ dendriteCfg.Global.ServerName,
relayingEnabled,
)
}
diff --git a/relayapi/relayapi_test.go b/relayapi/relayapi_test.go
index e8120309..8973c8cc 100644
--- a/relayapi/relayapi_test.go
+++ b/relayapi/relayapi_test.go
@@ -21,10 +21,13 @@ import (
"net/http"
"net/http/httptest"
"testing"
+ "time"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/relayapi"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
@@ -34,38 +37,38 @@ import (
func TestCreateNewRelayInternalAPI(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
-
- relayAPI := relayapi.NewRelayInternalAPI(base, nil, nil, nil, nil, true, caches)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ relayAPI := relayapi.NewRelayInternalAPI(cfg, cm, nil, nil, nil, nil, true, caches)
assert.NotNil(t, relayAPI)
})
}
func TestCreateRelayInternalInvalidDatabasePanics(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
if dbType == test.DBTypeSQLite {
- base.Cfg.RelayAPI.Database.ConnectionString = "file:"
+ cfg.RelayAPI.Database.ConnectionString = "file:"
} else {
- base.Cfg.RelayAPI.Database.ConnectionString = "test"
+ cfg.RelayAPI.Database.ConnectionString = "test"
}
defer close()
-
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
assert.Panics(t, func() {
- relayapi.NewRelayInternalAPI(base, nil, nil, nil, nil, true, nil)
+ relayapi.NewRelayInternalAPI(cfg, cm, nil, nil, nil, nil, true, nil)
})
})
}
func TestCreateInvalidRelayPublicRoutesPanics(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, _, close := testrig.CreateConfig(t, dbType)
defer close()
-
+ routers := httputil.NewRouters()
assert.Panics(t, func() {
- relayapi.AddPublicRoutes(base, nil, nil)
+ relayapi.AddPublicRoutes(routers, cfg, nil, nil)
})
})
}
@@ -107,16 +110,19 @@ func createSendRelayTxnHTTPRequest(serverName gomatrixserverlib.ServerName, txnI
func TestCreateRelayPublicRoutes(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
+ routers := httputil.NewRouters()
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
- relayAPI := relayapi.NewRelayInternalAPI(base, nil, nil, nil, nil, true, caches)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+
+ relayAPI := relayapi.NewRelayInternalAPI(cfg, cm, nil, nil, nil, nil, true, caches)
assert.NotNil(t, relayAPI)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
- relayapi.AddPublicRoutes(base, keyRing, relayAPI)
+ relayapi.AddPublicRoutes(routers, cfg, keyRing, relayAPI)
testCases := []struct {
name string
@@ -125,29 +131,29 @@ func TestCreateRelayPublicRoutes(t *testing.T) {
}{
{
name: "relay_txn invalid user id",
- req: createGetRelayTxnHTTPRequest(base.Cfg.Global.ServerName, "user:local"),
+ req: createGetRelayTxnHTTPRequest(cfg.Global.ServerName, "user:local"),
wantCode: 400,
},
{
name: "relay_txn valid user id",
- req: createGetRelayTxnHTTPRequest(base.Cfg.Global.ServerName, "@user:local"),
+ req: createGetRelayTxnHTTPRequest(cfg.Global.ServerName, "@user:local"),
wantCode: 200,
},
{
name: "send_relay invalid user id",
- req: createSendRelayTxnHTTPRequest(base.Cfg.Global.ServerName, "123", "user:local"),
+ req: createSendRelayTxnHTTPRequest(cfg.Global.ServerName, "123", "user:local"),
wantCode: 400,
},
{
name: "send_relay valid user id",
- req: createSendRelayTxnHTTPRequest(base.Cfg.Global.ServerName, "123", "@user:local"),
+ req: createSendRelayTxnHTTPRequest(cfg.Global.ServerName, "123", "@user:local"),
wantCode: 200,
},
}
for _, tc := range testCases {
w := httptest.NewRecorder()
- base.Routers.Federation.ServeHTTP(w, tc.req)
+ routers.Federation.ServeHTTP(w, tc.req)
if w.Code != tc.wantCode {
t.Fatalf("%s: got HTTP %d want %d", tc.name, w.Code, tc.wantCode)
}
@@ -157,16 +163,19 @@ func TestCreateRelayPublicRoutes(t *testing.T) {
func TestDisableRelayPublicRoutes(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
+ routers := httputil.NewRouters()
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
- relayAPI := relayapi.NewRelayInternalAPI(base, nil, nil, nil, nil, false, caches)
+ relayAPI := relayapi.NewRelayInternalAPI(cfg, cm, nil, nil, nil, nil, false, caches)
assert.NotNil(t, relayAPI)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
- relayapi.AddPublicRoutes(base, keyRing, relayAPI)
+ relayapi.AddPublicRoutes(routers, cfg, keyRing, relayAPI)
testCases := []struct {
name string
@@ -175,19 +184,19 @@ func TestDisableRelayPublicRoutes(t *testing.T) {
}{
{
name: "relay_txn valid user id",
- req: createGetRelayTxnHTTPRequest(base.Cfg.Global.ServerName, "@user:local"),
+ req: createGetRelayTxnHTTPRequest(cfg.Global.ServerName, "@user:local"),
wantCode: 404,
},
{
name: "send_relay valid user id",
- req: createSendRelayTxnHTTPRequest(base.Cfg.Global.ServerName, "123", "@user:local"),
+ req: createSendRelayTxnHTTPRequest(cfg.Global.ServerName, "123", "@user:local"),
wantCode: 404,
},
}
for _, tc := range testCases {
w := httptest.NewRecorder()
- base.Routers.Federation.ServeHTTP(w, tc.req)
+ routers.Federation.ServeHTTP(w, tc.req)
if w.Code != tc.wantCode {
t.Fatalf("%s: got HTTP %d want %d", tc.name, w.Code, tc.wantCode)
}
diff --git a/roomserver/internal/alias.go b/roomserver/internal/alias.go
index fc61b7f4..94b8b16c 100644
--- a/roomserver/internal/alias.go
+++ b/roomserver/internal/alias.go
@@ -117,7 +117,7 @@ func (r *RoomserverInternalAPI) RemoveRoomAlias(
request *api.RemoveRoomAliasRequest,
response *api.RemoveRoomAliasResponse,
) error {
- _, virtualHost, err := r.Cfg.Matrix.SplitLocalID('@', request.UserID)
+ _, virtualHost, err := r.Cfg.Global.SplitLocalID('@', request.UserID)
if err != nil {
return err
}
@@ -175,12 +175,12 @@ func (r *RoomserverInternalAPI) RemoveRoomAlias(
sender = ev.Sender()
}
- _, senderDomain, err := r.Cfg.Matrix.SplitLocalID('@', sender)
+ _, senderDomain, err := r.Cfg.Global.SplitLocalID('@', sender)
if err != nil {
return err
}
- identity, err := r.Cfg.Matrix.SigningIdentityFor(senderDomain)
+ identity, err := r.Cfg.Global.SigningIdentityFor(senderDomain)
if err != nil {
return err
}
@@ -206,7 +206,7 @@ func (r *RoomserverInternalAPI) RemoveRoomAlias(
return err
}
- newEvent, err := eventutil.BuildEvent(ctx, builder, r.Cfg.Matrix, identity, time.Now(), &eventsNeeded, stateRes)
+ newEvent, err := eventutil.BuildEvent(ctx, builder, &r.Cfg.Global, identity, time.Now(), &eventsNeeded, stateRes)
if err != nil {
return err
}
diff --git a/roomserver/internal/api.go b/roomserver/internal/api.go
index 2e987d68..7ca3675d 100644
--- a/roomserver/internal/api.go
+++ b/roomserver/internal/api.go
@@ -18,7 +18,6 @@ import (
"github.com/matrix-org/dendrite/roomserver/internal/query"
"github.com/matrix-org/dendrite/roomserver/producers"
"github.com/matrix-org/dendrite/roomserver/storage"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
@@ -41,9 +40,8 @@ type RoomserverInternalAPI struct {
*perform.Upgrader
*perform.Admin
ProcessContext *process.ProcessContext
- Base *base.BaseDendrite
DB storage.Database
- Cfg *config.RoomServer
+ Cfg *config.Dendrite
Cache caching.RoomServerCaches
ServerName gomatrixserverlib.ServerName
KeyRing gomatrixserverlib.JSONVerifier
@@ -56,43 +54,44 @@ type RoomserverInternalAPI struct {
InputRoomEventTopic string // JetStream topic for new input room events
OutputProducer *producers.RoomEventProducer
PerspectiveServerNames []gomatrixserverlib.ServerName
+ enableMetrics bool
}
func NewRoomserverAPI(
- base *base.BaseDendrite, roomserverDB storage.Database,
- js nats.JetStreamContext, nc *nats.Conn, caches caching.RoomServerCaches,
+ processContext *process.ProcessContext, dendriteCfg *config.Dendrite, roomserverDB storage.Database,
+ js nats.JetStreamContext, nc *nats.Conn, caches caching.RoomServerCaches, enableMetrics bool,
) *RoomserverInternalAPI {
var perspectiveServerNames []gomatrixserverlib.ServerName
- for _, kp := range base.Cfg.FederationAPI.KeyPerspectives {
+ for _, kp := range dendriteCfg.FederationAPI.KeyPerspectives {
perspectiveServerNames = append(perspectiveServerNames, kp.ServerName)
}
serverACLs := acls.NewServerACLs(roomserverDB)
producer := &producers.RoomEventProducer{
- Topic: string(base.Cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent)),
+ Topic: string(dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent)),
JetStream: js,
ACLs: serverACLs,
}
a := &RoomserverInternalAPI{
- ProcessContext: base.ProcessContext,
+ ProcessContext: processContext,
DB: roomserverDB,
- Base: base,
- Cfg: &base.Cfg.RoomServer,
+ Cfg: dendriteCfg,
Cache: caches,
- ServerName: base.Cfg.Global.ServerName,
+ ServerName: dendriteCfg.Global.ServerName,
PerspectiveServerNames: perspectiveServerNames,
- InputRoomEventTopic: base.Cfg.Global.JetStream.Prefixed(jetstream.InputRoomEvent),
+ InputRoomEventTopic: dendriteCfg.Global.JetStream.Prefixed(jetstream.InputRoomEvent),
OutputProducer: producer,
JetStream: js,
NATSClient: nc,
- Durable: base.Cfg.Global.JetStream.Durable("RoomserverInputConsumer"),
+ Durable: dendriteCfg.Global.JetStream.Durable("RoomserverInputConsumer"),
ServerACLs: serverACLs,
Queryer: &query.Queryer{
DB: roomserverDB,
Cache: caches,
- IsLocalServerName: base.Cfg.Global.IsLocalServerName,
+ IsLocalServerName: dendriteCfg.Global.IsLocalServerName,
ServerACLs: serverACLs,
},
+ enableMetrics: enableMetrics,
// perform-er structs get initialised when we have a federation sender to use
}
return a
@@ -105,15 +104,14 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
r.fsAPI = fsAPI
r.KeyRing = keyRing
- identity, err := r.Cfg.Matrix.SigningIdentityFor(r.ServerName)
+ identity, err := r.Cfg.Global.SigningIdentityFor(r.ServerName)
if err != nil {
logrus.Panic(err)
}
r.Inputer = &input.Inputer{
- Cfg: &r.Base.Cfg.RoomServer,
- Base: r.Base,
- ProcessContext: r.Base.ProcessContext,
+ Cfg: &r.Cfg.RoomServer,
+ ProcessContext: r.ProcessContext,
DB: r.DB,
InputRoomEventTopic: r.InputRoomEventTopic,
OutputProducer: r.OutputProducer,
@@ -129,12 +127,12 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
}
r.Inviter = &perform.Inviter{
DB: r.DB,
- Cfg: r.Cfg,
+ Cfg: &r.Cfg.RoomServer,
FSAPI: r.fsAPI,
Inputer: r.Inputer,
}
r.Joiner = &perform.Joiner{
- Cfg: r.Cfg,
+ Cfg: &r.Cfg.RoomServer,
DB: r.DB,
FSAPI: r.fsAPI,
RSAPI: r,
@@ -143,7 +141,7 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
}
r.Peeker = &perform.Peeker{
ServerName: r.ServerName,
- Cfg: r.Cfg,
+ Cfg: &r.Cfg.RoomServer,
DB: r.DB,
FSAPI: r.fsAPI,
Inputer: r.Inputer,
@@ -154,12 +152,12 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
}
r.Unpeeker = &perform.Unpeeker{
ServerName: r.ServerName,
- Cfg: r.Cfg,
+ Cfg: &r.Cfg.RoomServer,
FSAPI: r.fsAPI,
Inputer: r.Inputer,
}
r.Leaver = &perform.Leaver{
- Cfg: r.Cfg,
+ Cfg: &r.Cfg.RoomServer,
DB: r.DB,
FSAPI: r.fsAPI,
Inputer: r.Inputer,
@@ -168,7 +166,7 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
DB: r.DB,
}
r.Backfiller = &perform.Backfiller{
- IsLocalServerName: r.Cfg.Matrix.IsLocalServerName,
+ IsLocalServerName: r.Cfg.Global.IsLocalServerName,
DB: r.DB,
FSAPI: r.fsAPI,
KeyRing: r.KeyRing,
@@ -181,12 +179,12 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
DB: r.DB,
}
r.Upgrader = &perform.Upgrader{
- Cfg: r.Cfg,
+ Cfg: &r.Cfg.RoomServer,
URSAPI: r,
}
r.Admin = &perform.Admin{
DB: r.DB,
- Cfg: r.Cfg,
+ Cfg: &r.Cfg.RoomServer,
Inputer: r.Inputer,
Queryer: r.Queryer,
Leaver: r.Leaver,
diff --git a/roomserver/internal/helpers/helpers_test.go b/roomserver/internal/helpers/helpers_test.go
index 03a8bf57..dd74b844 100644
--- a/roomserver/internal/helpers/helpers_test.go
+++ b/roomserver/internal/helpers/helpers_test.go
@@ -19,7 +19,7 @@ import (
func mustCreateDatabase(t *testing.T, dbType test.DBType) (storage.Database, func()) {
conStr, close := test.PrepareDBConnectionString(t, dbType)
caches := caching.NewRistrettoCache(8*1024*1024, time.Hour, caching.DisableMetrics)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.Open(context.Background(), cm, &config.DatabaseOptions{ConnectionString: config.DataSource(conStr)}, caches)
if err != nil {
t.Fatalf("failed to create Database: %v", err)
diff --git a/roomserver/internal/input/input.go b/roomserver/internal/input/input.go
index 2ec19f01..cc0c673d 100644
--- a/roomserver/internal/input/input.go
+++ b/roomserver/internal/input/input.go
@@ -39,7 +39,6 @@ import (
"github.com/matrix-org/dendrite/roomserver/producers"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
@@ -74,7 +73,6 @@ import (
// or C.
type Inputer struct {
Cfg *config.RoomServer
- Base *base.BaseDendrite
ProcessContext *process.ProcessContext
DB storage.RoomDatabase
NATSClient *nats.Conn
@@ -89,8 +87,9 @@ type Inputer struct {
OutputProducer *producers.RoomEventProducer
workers sync.Map // room ID -> *worker
- Queryer *query.Queryer
- UserAPI userapi.RoomserverUserAPI
+ Queryer *query.Queryer
+ UserAPI userapi.RoomserverUserAPI
+ enableMetrics bool
}
// If a room consumer is inactive for a while then we will allow NATS
@@ -177,7 +176,7 @@ func (r *Inputer) startWorkerForRoom(roomID string) {
// will look to see if we have a worker for that room which has its
// own consumer. If we don't, we'll start one.
func (r *Inputer) Start() error {
- if r.Base.EnableMetrics {
+ if r.enableMetrics {
prometheus.MustRegister(roomserverInputBackpressure, processRoomEventDuration)
}
_, err := r.JetStream.Subscribe(
diff --git a/roomserver/internal/input/input_test.go b/roomserver/internal/input/input_test.go
index 555ec9c6..51c50c37 100644
--- a/roomserver/internal/input/input_test.go
+++ b/roomserver/internal/input/input_test.go
@@ -2,84 +2,69 @@ package input_test
import (
"context"
- "os"
"testing"
"time"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/input"
- "github.com/matrix-org/dendrite/roomserver/storage"
- "github.com/matrix-org/dendrite/setup/base"
- "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/gomatrixserverlib"
- "github.com/nats-io/nats.go"
)
-var js nats.JetStreamContext
-var jc *nats.Conn
+func TestSingleTransactionOnInput(t *testing.T) {
+ test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ defer close()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
-func TestMain(m *testing.M) {
- var b *base.BaseDendrite
- b, js, jc = testrig.Base(nil)
- code := m.Run()
- b.ShutdownDendrite()
- b.WaitForComponentsToFinish()
- os.Exit(code)
-}
+ natsInstance := &jetstream.NATSInstance{}
+ js, jc := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ caches := caching.NewRistrettoCache(8*1024*1024, time.Hour, caching.DisableMetrics)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, natsInstance, caches, caching.DisableMetrics)
+ rsAPI.SetFederationAPI(nil, nil)
-func TestSingleTransactionOnInput(t *testing.T) {
- deadline, _ := t.Deadline()
- if max := time.Now().Add(time.Second * 3); deadline.After(max) {
- deadline = max
- }
- ctx, cancel := context.WithDeadline(context.Background(), deadline)
- defer cancel()
+ deadline, _ := t.Deadline()
+ if max := time.Now().Add(time.Second * 3); deadline.Before(max) {
+ deadline = max
+ }
+ ctx, cancel := context.WithDeadline(processCtx.Context(), deadline)
+ defer cancel()
+
+ event, err := gomatrixserverlib.NewEventFromTrustedJSON(
+ []byte(`{"auth_events":[],"content":{"creator":"@neilalexander:dendrite.matrix.org","room_version":"6"},"depth":1,"hashes":{"sha256":"jqOqdNEH5r0NiN3xJtj0u5XUVmRqq9YvGbki1wxxuuM"},"origin":"dendrite.matrix.org","origin_server_ts":1644595362726,"prev_events":[],"prev_state":[],"room_id":"!jSZZRknA6GkTBXNP:dendrite.matrix.org","sender":"@neilalexander:dendrite.matrix.org","signatures":{"dendrite.matrix.org":{"ed25519:6jB2aB":"bsQXO1wketf1OSe9xlndDIWe71W9KIundc6rBw4KEZdGPW7x4Tv4zDWWvbxDsG64sS2IPWfIm+J0OOozbrWIDw"}},"state_key":"","type":"m.room.create"}`),
+ false, gomatrixserverlib.RoomVersionV6,
+ )
+ if err != nil {
+ t.Fatal(err)
+ }
+ in := api.InputRoomEvent{
+ Kind: api.KindOutlier, // don't panic if we generate an output event
+ Event: event.Headered(gomatrixserverlib.RoomVersionV6),
+ }
- event, err := gomatrixserverlib.NewEventFromTrustedJSON(
- []byte(`{"auth_events":[],"content":{"creator":"@neilalexander:dendrite.matrix.org","room_version":"6"},"depth":1,"hashes":{"sha256":"jqOqdNEH5r0NiN3xJtj0u5XUVmRqq9YvGbki1wxxuuM"},"origin":"dendrite.matrix.org","origin_server_ts":1644595362726,"prev_events":[],"prev_state":[],"room_id":"!jSZZRknA6GkTBXNP:dendrite.matrix.org","sender":"@neilalexander:dendrite.matrix.org","signatures":{"dendrite.matrix.org":{"ed25519:6jB2aB":"bsQXO1wketf1OSe9xlndDIWe71W9KIundc6rBw4KEZdGPW7x4Tv4zDWWvbxDsG64sS2IPWfIm+J0OOozbrWIDw"}},"state_key":"","type":"m.room.create"}`),
- false, gomatrixserverlib.RoomVersionV6,
- )
- if err != nil {
- t.Fatal(err)
- }
- in := api.InputRoomEvent{
- Kind: api.KindOutlier, // don't panic if we generate an output event
- Event: event.Headered(gomatrixserverlib.RoomVersionV6),
- }
- cm := sqlutil.NewConnectionManager()
- db, err := storage.Open(
- context.Background(), cm,
- &config.DatabaseOptions{
- ConnectionString: "",
- MaxOpenConnections: 1,
- MaxIdleConnections: 1,
- },
- caching.NewRistrettoCache(8*1024*1024, time.Hour, caching.DisableMetrics),
- )
- if err != nil {
- t.Logf("PostgreSQL not available (%s), skipping", err)
- t.SkipNow()
- }
- inputter := &input.Inputer{
- DB: db,
- JetStream: js,
- NATSClient: jc,
- }
- res := &api.InputRoomEventsResponse{}
- inputter.InputRoomEvents(
- ctx,
- &api.InputRoomEventsRequest{
- InputRoomEvents: []api.InputRoomEvent{in},
- Asynchronous: false,
- },
- res,
- )
- // If we fail here then it's because we've hit the test deadline,
- // so we probably deadlocked
- if err := res.Err(); err != nil {
- t.Fatal(err)
- }
+ inputter := &input.Inputer{
+ JetStream: js,
+ NATSClient: jc,
+ Cfg: &cfg.RoomServer,
+ }
+ res := &api.InputRoomEventsResponse{}
+ inputter.InputRoomEvents(
+ ctx,
+ &api.InputRoomEventsRequest{
+ InputRoomEvents: []api.InputRoomEvent{in},
+ Asynchronous: false,
+ },
+ res,
+ )
+ // If we fail here then it's because we've hit the test deadline,
+ // so we probably deadlocked
+ if err := res.Err(); err != nil {
+ t.Fatal(err)
+ }
+ })
}
diff --git a/roomserver/roomserver.go b/roomserver/roomserver.go
index 1c55423e..4685f474 100644
--- a/roomserver/roomserver.go
+++ b/roomserver/roomserver.go
@@ -16,29 +16,34 @@ package roomserver
import (
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal"
"github.com/matrix-org/dendrite/roomserver/storage"
- "github.com/matrix-org/dendrite/setup/base"
)
// NewInternalAPI returns a concrete implementation of the internal API.
func NewInternalAPI(
- base *base.BaseDendrite,
+ processContext *process.ProcessContext,
+ cfg *config.Dendrite,
+ cm sqlutil.Connections,
+ natsInstance *jetstream.NATSInstance,
caches caching.RoomServerCaches,
+ enableMetrics bool,
) api.RoomserverInternalAPI {
- cfg := &base.Cfg.RoomServer
-
- roomserverDB, err := storage.Open(base.ProcessContext.Context(), base.ConnectionManager, &cfg.Database, caches)
+ roomserverDB, err := storage.Open(processContext.Context(), cm, &cfg.RoomServer.Database, caches)
if err != nil {
logrus.WithError(err).Panicf("failed to connect to room server db")
}
- js, nc := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
+ js, nc := natsInstance.Prepare(processContext, &cfg.Global.JetStream)
return internal.NewRoomserverAPI(
- base, roomserverDB, js, nc, caches,
+ processContext, cfg, roomserverDB, js, nc, caches, enableMetrics,
)
}
diff --git a/roomserver/roomserver_test.go b/roomserver/roomserver_test.go
index 1b0b3155..729da15b 100644
--- a/roomserver/roomserver_test.go
+++ b/roomserver/roomserver_test.go
@@ -8,11 +8,12 @@ import (
"time"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/stretchr/testify/assert"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/types"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/userapi"
userAPI "github.com/matrix-org/dendrite/userapi/api"
@@ -30,23 +31,14 @@ import (
"github.com/matrix-org/dendrite/test/testrig"
)
-func mustCreateDatabase(t *testing.T, dbType test.DBType) (*base.BaseDendrite, storage.Database, func()) {
- t.Helper()
- base, close := testrig.CreateBaseDendrite(t, dbType)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- db, err := storage.Open(base.ProcessContext.Context(), base.ConnectionManager, &base.Cfg.RoomServer.Database, caches)
- if err != nil {
- t.Fatalf("failed to create Database: %v", err)
- }
- return base, db, close
-}
-
func TestUsers(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
// SetFederationAPI starts the room event input consumer
rsAPI.SetFederationAPI(nil, nil)
@@ -55,7 +47,7 @@ func TestUsers(t *testing.T) {
})
t.Run("kick users", func(t *testing.T) {
- usrAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ usrAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
rsAPI.SetUserAPI(usrAPI)
testKickUsers(t, rsAPI, usrAPI)
})
@@ -181,11 +173,13 @@ func Test_QueryLeftUsers(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, _, close := mustCreateDatabase(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
// SetFederationAPI starts the room event input consumer
rsAPI.SetFederationAPI(nil, nil)
// Create the room
@@ -229,30 +223,35 @@ func TestPurgeRoom(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, db, close := mustCreateDatabase(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ natsInstance := jetstream.NATSInstance{}
defer close()
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ db, err := storage.Open(processCtx.Context(), cm, &cfg.RoomServer.Database, caches)
+ if err != nil {
+ t.Fatal(err)
+ }
+ jsCtx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsCtx, &cfg.Global.JetStream)
- jsCtx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsCtx, &base.Cfg.Global.JetStream)
-
- fedClient := base.CreateFederationClient()
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
- userAPI := userapi.NewInternalAPI(base, rsAPI, nil)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
+ userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil)
// this starts the JetStream consumers
- syncapi.AddPublicRoutes(base, userAPI, rsAPI, caches)
- federationapi.NewInternalAPI(base, fedClient, rsAPI, caches, nil, true)
+ syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, userAPI, rsAPI, caches, caching.DisableMetrics)
+ federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(nil, nil)
// Create the room
- if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
+ if err = api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// some dummy entries to validate after purging
publishResp := &api.PerformPublishResponse{}
- if err := rsAPI.PerformPublish(ctx, &api.PerformPublishRequest{RoomID: room.ID, Visibility: "public"}, publishResp); err != nil {
+ if err = rsAPI.PerformPublish(ctx, &api.PerformPublishRequest{RoomID: room.ID, Visibility: "public"}, publishResp); err != nil {
t.Fatal(err)
}
if publishResp.Error != nil {
@@ -340,7 +339,7 @@ func TestPurgeRoom(t *testing.T) {
t.Fatalf("test timed out after %s", timeout)
}
sum = 0
- consumerCh := jsCtx.Consumers(base.Cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent))
+ consumerCh := jsCtx.Consumers(cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent))
for x := range consumerCh {
sum += x.NumAckPending
}
@@ -508,8 +507,14 @@ func TestRedaction(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- _, db, close := mustCreateDatabase(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
defer close()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ db, err := storage.Open(processCtx.Context(), cm, &cfg.RoomServer.Database, caches)
+ if err != nil {
+ t.Fatal(err)
+ }
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
diff --git a/roomserver/storage/shared/storage_test.go b/roomserver/storage/shared/storage_test.go
index 684e80b8..941e8480 100644
--- a/roomserver/storage/shared/storage_test.go
+++ b/roomserver/storage/shared/storage_test.go
@@ -15,14 +15,12 @@ import (
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
- "github.com/matrix-org/dendrite/test/testrig"
)
func mustCreateRoomserverDatabase(t *testing.T, dbType test.DBType) (*shared.Database, func()) {
t.Helper()
connStr, clearDB := test.PrepareDBConnectionString(t, dbType)
- base, _, _ := testrig.Base(nil)
dbOpts := &config.DatabaseOptions{ConnectionString: config.DataSource(connStr)}
db, err := sqlutil.Open(dbOpts, sqlutil.NewExclusiveWriter())
@@ -61,8 +59,6 @@ func mustCreateRoomserverDatabase(t *testing.T, dbType test.DBType) (*shared.Dat
Writer: sqlutil.NewExclusiveWriter(),
Cache: cache,
}, func() {
- err := base.Close()
- assert.NoError(t, err)
clearDB()
err = db.Close()
assert.NoError(t, err)
diff --git a/setup/base/base.go b/setup/base/base.go
index 8c9b06d0..e5fd6fed 100644
--- a/setup/base/base.go
+++ b/setup/base/base.go
@@ -22,30 +22,24 @@ import (
"errors"
"fmt"
"html/template"
- "io"
"io/fs"
"net"
"net/http"
_ "net/http/pprof"
"os"
"os/signal"
- "sync"
"syscall"
"time"
- "github.com/getsentry/sentry-go"
sentryhttp "github.com/getsentry/sentry-go/http"
- "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/gomatrixserverlib"
"github.com/prometheus/client_golang/prometheus/promhttp"
"go.uber.org/atomic"
- "github.com/matrix-org/dendrite/internal"
- "github.com/matrix-org/dendrite/internal/httputil"
- "github.com/matrix-org/dendrite/internal/sqlutil"
-
"github.com/gorilla/mux"
"github.com/kardianos/minwinsvc"
+ "github.com/matrix-org/dendrite/internal"
+ "github.com/matrix-org/dendrite/internal/httputil"
"github.com/sirupsen/logrus"
@@ -56,154 +50,22 @@ import (
//go:embed static/*.gotmpl
var staticContent embed.FS
-// BaseDendrite is a base for creating new instances of dendrite. It parses
-// command line flags and config, and exposes methods for creating various
-// resources. All errors are handled by logging then exiting, so all methods
-// should only be used during start up.
-// Must be closed when shutting down.
-type BaseDendrite struct {
- *process.ProcessContext
- tracerCloser io.Closer
- Routers httputil.Routers
- NATS *jetstream.NATSInstance
- Cfg *config.Dendrite
- DNSCache *gomatrixserverlib.DNSCache
- ConnectionManager sqlutil.Connections
- EnableMetrics bool
- startupLock sync.Mutex
-}
-
const HTTPServerTimeout = time.Minute * 5
-type BaseDendriteOptions int
-
-const (
- DisableMetrics BaseDendriteOptions = iota
-)
-
-// NewBaseDendrite creates a new instance to be used by a component.
-func NewBaseDendrite(cfg *config.Dendrite, options ...BaseDendriteOptions) *BaseDendrite {
- platformSanityChecks()
- enableMetrics := true
- for _, opt := range options {
- switch opt {
- case DisableMetrics:
- enableMetrics = false
- }
- }
-
- configErrors := &config.ConfigErrors{}
- cfg.Verify(configErrors)
- if len(*configErrors) > 0 {
- for _, err := range *configErrors {
- logrus.Errorf("Configuration error: %s", err)
- }
- logrus.Fatalf("Failed to start due to configuration errors")
- }
-
- internal.SetupStdLogging()
- internal.SetupHookLogging(cfg.Logging)
- internal.SetupPprof()
-
- logrus.Infof("Dendrite version %s", internal.VersionString())
-
- if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
- logrus.Warn("Open registration is enabled")
- }
-
- closer, err := cfg.SetupTracing()
- if err != nil {
- logrus.WithError(err).Panicf("failed to start opentracing")
- }
-
- if cfg.Global.Sentry.Enabled {
- logrus.Info("Setting up Sentry for debugging...")
- err = sentry.Init(sentry.ClientOptions{
- Dsn: cfg.Global.Sentry.DSN,
- Environment: cfg.Global.Sentry.Environment,
- Debug: true,
- ServerName: string(cfg.Global.ServerName),
- Release: "dendrite@" + internal.VersionString(),
- AttachStacktrace: true,
- })
- if err != nil {
- logrus.WithError(err).Panic("failed to start Sentry")
- }
- }
-
- var dnsCache *gomatrixserverlib.DNSCache
- if cfg.Global.DNSCache.Enabled {
- dnsCache = gomatrixserverlib.NewDNSCache(
- cfg.Global.DNSCache.CacheSize,
- cfg.Global.DNSCache.CacheLifetime,
- )
- logrus.Infof(
- "DNS cache enabled (size %d, lifetime %s)",
- cfg.Global.DNSCache.CacheSize,
- cfg.Global.DNSCache.CacheLifetime,
- )
- }
-
- // If we're in monolith mode, we'll set up a global pool of database
- // connections. A component is welcome to use this pool if they don't
- // have a separate database config of their own.
- cm := sqlutil.NewConnectionManager()
- if cfg.Global.DatabaseOptions.ConnectionString != "" {
- if cfg.Global.DatabaseOptions.ConnectionString.IsSQLite() {
- logrus.Panic("Using a global database connection pool is not supported with SQLite databases")
- }
- _, _, err := cm.Connection(&cfg.Global.DatabaseOptions)
- if err != nil {
- logrus.WithError(err).Panic("Failed to set up global database connections")
- }
- logrus.Debug("Using global database connection pool")
- }
-
- // Ideally we would only use SkipClean on routes which we know can allow '/' but due to
- // https://github.com/gorilla/mux/issues/460 we have to attach this at the top router.
- // When used in conjunction with UseEncodedPath() we get the behaviour we want when parsing
- // path parameters:
- // /foo/bar%2Fbaz == [foo, bar%2Fbaz] (from UseEncodedPath)
- // /foo/bar%2F%2Fbaz == [foo, bar%2F%2Fbaz] (from SkipClean)
- // In particular, rooms v3 event IDs are not urlsafe and can include '/' and because they
- // are randomly generated it results in flakey tests.
- // We need to be careful with media APIs if they read from a filesystem to make sure they
- // are not inadvertently reading paths without cleaning, else this could introduce a
- // directory traversal attack e.g /../../../etc/passwd
-
- return &BaseDendrite{
- ProcessContext: process.NewProcessContext(),
- tracerCloser: closer,
- Cfg: cfg,
- DNSCache: dnsCache,
- Routers: httputil.NewRouters(),
- NATS: &jetstream.NATSInstance{},
- ConnectionManager: cm,
- EnableMetrics: enableMetrics,
- }
-}
-
-// Close implements io.Closer
-func (b *BaseDendrite) Close() error {
- b.ProcessContext.ShutdownDendrite()
- b.ProcessContext.WaitForShutdown()
- return b.tracerCloser.Close()
-}
-
// CreateClient creates a new client (normally used for media fetch requests).
// Should only be called once per component.
-func (b *BaseDendrite) CreateClient() *gomatrixserverlib.Client {
- if b.Cfg.Global.DisableFederation {
+func CreateClient(cfg *config.Dendrite, dnsCache *gomatrixserverlib.DNSCache) *gomatrixserverlib.Client {
+ if cfg.Global.DisableFederation {
return gomatrixserverlib.NewClient(
gomatrixserverlib.WithTransport(noOpHTTPTransport),
)
}
opts := []gomatrixserverlib.ClientOption{
- gomatrixserverlib.WithSkipVerify(b.Cfg.FederationAPI.DisableTLSValidation),
+ gomatrixserverlib.WithSkipVerify(cfg.FederationAPI.DisableTLSValidation),
gomatrixserverlib.WithWellKnownSRVLookups(true),
}
- if b.Cfg.Global.DNSCache.Enabled {
- opts = append(opts, gomatrixserverlib.WithDNSCache(b.DNSCache))
+ if cfg.Global.DNSCache.Enabled && dnsCache != nil {
+ opts = append(opts, gomatrixserverlib.WithDNSCache(dnsCache))
}
client := gomatrixserverlib.NewClient(opts...)
client.SetUserAgent(fmt.Sprintf("Dendrite/%s", internal.VersionString()))
@@ -212,20 +74,20 @@ func (b *BaseDendrite) CreateClient() *gomatrixserverlib.Client {
// CreateFederationClient creates a new federation client. Should only be called
// once per component.
-func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationClient {
- identities := b.Cfg.Global.SigningIdentities()
- if b.Cfg.Global.DisableFederation {
+func CreateFederationClient(cfg *config.Dendrite, dnsCache *gomatrixserverlib.DNSCache) *gomatrixserverlib.FederationClient {
+ identities := cfg.Global.SigningIdentities()
+ if cfg.Global.DisableFederation {
return gomatrixserverlib.NewFederationClient(
identities, gomatrixserverlib.WithTransport(noOpHTTPTransport),
)
}
opts := []gomatrixserverlib.ClientOption{
gomatrixserverlib.WithTimeout(time.Minute * 5),
- gomatrixserverlib.WithSkipVerify(b.Cfg.FederationAPI.DisableTLSValidation),
- gomatrixserverlib.WithKeepAlives(!b.Cfg.FederationAPI.DisableHTTPKeepalives),
+ gomatrixserverlib.WithSkipVerify(cfg.FederationAPI.DisableTLSValidation),
+ gomatrixserverlib.WithKeepAlives(!cfg.FederationAPI.DisableHTTPKeepalives),
}
- if b.Cfg.Global.DNSCache.Enabled {
- opts = append(opts, gomatrixserverlib.WithDNSCache(b.DNSCache))
+ if cfg.Global.DNSCache.Enabled {
+ opts = append(opts, gomatrixserverlib.WithDNSCache(dnsCache))
}
client := gomatrixserverlib.NewFederationClient(
identities, opts...,
@@ -234,12 +96,12 @@ func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationCli
return client
}
-func (b *BaseDendrite) ConfigureAdminEndpoints() {
- b.Routers.DendriteAdmin.HandleFunc("/monitor/up", func(w http.ResponseWriter, r *http.Request) {
+func ConfigureAdminEndpoints(processContext *process.ProcessContext, routers httputil.Routers) {
+ routers.DendriteAdmin.HandleFunc("/monitor/up", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(200)
})
- b.Routers.DendriteAdmin.HandleFunc("/monitor/health", func(w http.ResponseWriter, r *http.Request) {
- if isDegraded, reasons := b.ProcessContext.IsDegraded(); isDegraded {
+ routers.DendriteAdmin.HandleFunc("/monitor/health", func(w http.ResponseWriter, r *http.Request) {
+ if isDegraded, reasons := processContext.IsDegraded(); isDegraded {
w.WriteHeader(503)
_ = json.NewEncoder(w).Encode(struct {
Warnings []string `json:"warnings"`
@@ -254,14 +116,13 @@ func (b *BaseDendrite) ConfigureAdminEndpoints() {
// SetupAndServeHTTP sets up the HTTP server to serve client & federation APIs
// and adds a prometheus handler under /_dendrite/metrics.
-func (b *BaseDendrite) SetupAndServeHTTP(
+func SetupAndServeHTTP(
+ processContext *process.ProcessContext,
+ cfg *config.Dendrite,
+ routers httputil.Routers,
externalHTTPAddr config.ServerAddress,
certFile, keyFile *string,
) {
- // Manually unlocked right before actually serving requests,
- // as we don't return from this method (defer doesn't work).
- b.startupLock.Lock()
-
externalRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
externalServ := &http.Server{
@@ -269,7 +130,7 @@ func (b *BaseDendrite) SetupAndServeHTTP(
WriteTimeout: HTTPServerTimeout,
Handler: externalRouter,
BaseContext: func(_ net.Listener) context.Context {
- return b.ProcessContext.Context()
+ return processContext.Context()
},
}
@@ -278,11 +139,11 @@ func (b *BaseDendrite) SetupAndServeHTTP(
http.Redirect(w, r, httputil.PublicStaticPath, http.StatusFound)
})
- if b.Cfg.Global.Metrics.Enabled {
- externalRouter.Handle("/metrics", httputil.WrapHandlerInBasicAuth(promhttp.Handler(), b.Cfg.Global.Metrics.BasicAuth))
+ if cfg.Global.Metrics.Enabled {
+ externalRouter.Handle("/metrics", httputil.WrapHandlerInBasicAuth(promhttp.Handler(), cfg.Global.Metrics.BasicAuth))
}
- b.ConfigureAdminEndpoints()
+ ConfigureAdminEndpoints(processContext, routers)
// Parse and execute the landing page template
tmpl := template.Must(template.ParseFS(staticContent, "static/*.gotmpl"))
@@ -293,38 +154,36 @@ func (b *BaseDendrite) SetupAndServeHTTP(
logrus.WithError(err).Fatal("failed to execute landing page template")
}
- b.Routers.Static.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+ routers.Static.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write(landingPage.Bytes())
})
var clientHandler http.Handler
- clientHandler = b.Routers.Client
- if b.Cfg.Global.Sentry.Enabled {
+ clientHandler = routers.Client
+ if cfg.Global.Sentry.Enabled {
sentryHandler := sentryhttp.New(sentryhttp.Options{
Repanic: true,
})
- clientHandler = sentryHandler.Handle(b.Routers.Client)
+ clientHandler = sentryHandler.Handle(routers.Client)
}
var federationHandler http.Handler
- federationHandler = b.Routers.Federation
- if b.Cfg.Global.Sentry.Enabled {
+ federationHandler = routers.Federation
+ if cfg.Global.Sentry.Enabled {
sentryHandler := sentryhttp.New(sentryhttp.Options{
Repanic: true,
})
- federationHandler = sentryHandler.Handle(b.Routers.Federation)
+ federationHandler = sentryHandler.Handle(routers.Federation)
}
- externalRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(b.Routers.DendriteAdmin)
+ externalRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
externalRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(clientHandler)
- if !b.Cfg.Global.DisableFederation {
- externalRouter.PathPrefix(httputil.PublicKeyPathPrefix).Handler(b.Routers.Keys)
+ if !cfg.Global.DisableFederation {
+ externalRouter.PathPrefix(httputil.PublicKeyPathPrefix).Handler(routers.Keys)
externalRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(federationHandler)
}
- externalRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(b.Routers.SynapseAdmin)
- externalRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(b.Routers.Media)
- externalRouter.PathPrefix(httputil.PublicWellKnownPrefix).Handler(b.Routers.WellKnown)
- externalRouter.PathPrefix(httputil.PublicStaticPath).Handler(b.Routers.Static)
-
- b.startupLock.Unlock()
+ externalRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
+ externalRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
+ externalRouter.PathPrefix(httputil.PublicWellKnownPrefix).Handler(routers.WellKnown)
+ externalRouter.PathPrefix(httputil.PublicStaticPath).Handler(routers.Static)
externalRouter.NotFoundHandler = httputil.NotFoundCORSHandler
externalRouter.MethodNotAllowedHandler = httputil.NotAllowedHandler
@@ -333,10 +192,10 @@ func (b *BaseDendrite) SetupAndServeHTTP(
go func() {
var externalShutdown atomic.Bool // RegisterOnShutdown can be called more than once
logrus.Infof("Starting external listener on %s", externalServ.Addr)
- b.ProcessContext.ComponentStarted()
+ processContext.ComponentStarted()
externalServ.RegisterOnShutdown(func() {
if externalShutdown.CompareAndSwap(false, true) {
- b.ProcessContext.ComponentFinished()
+ processContext.ComponentFinished()
logrus.Infof("Stopped external HTTP listener")
}
})
@@ -378,32 +237,27 @@ func (b *BaseDendrite) SetupAndServeHTTP(
}()
}
- minwinsvc.SetOnExit(b.ProcessContext.ShutdownDendrite)
- <-b.ProcessContext.WaitForShutdown()
+ minwinsvc.SetOnExit(processContext.ShutdownDendrite)
+ <-processContext.WaitForShutdown()
logrus.Infof("Stopping HTTP listeners")
_ = externalServ.Shutdown(context.Background())
logrus.Infof("Stopped HTTP listeners")
}
-func (b *BaseDendrite) WaitForShutdown() {
+func WaitForShutdown(processCtx *process.ProcessContext) {
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
select {
case <-sigs:
- case <-b.ProcessContext.WaitForShutdown():
+ case <-processCtx.WaitForShutdown():
}
signal.Reset(syscall.SIGINT, syscall.SIGTERM)
logrus.Warnf("Shutdown signal received")
- b.ProcessContext.ShutdownDendrite()
- b.ProcessContext.WaitForComponentsToFinish()
- if b.Cfg.Global.Sentry.Enabled {
- if !sentry.Flush(time.Second * 5) {
- logrus.Warnf("failed to flush all Sentry events!")
- }
- }
+ processCtx.ShutdownDendrite()
+ processCtx.WaitForComponentsToFinish()
logrus.Warnf("Dendrite is exiting now")
}
diff --git a/setup/base/base_test.go b/setup/base/base_test.go
index aa406db2..bba967b9 100644
--- a/setup/base/base_test.go
+++ b/setup/base/base_test.go
@@ -13,8 +13,10 @@ import (
"time"
"github.com/matrix-org/dendrite/internal"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ basepkg "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
- "github.com/matrix-org/dendrite/test/testrig"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/stretchr/testify/assert"
)
@@ -30,8 +32,10 @@ func TestLandingPage_Tcp(t *testing.T) {
})
assert.NoError(t, err)
- b, _, _ := testrig.Base(nil)
- defer b.Close()
+ processCtx := process.NewProcessContext()
+ routers := httputil.NewRouters()
+ cfg := config.Dendrite{}
+ cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
// hack: create a server and close it immediately, just to get a random port assigned
s := httptest.NewServer(nil)
@@ -40,7 +44,7 @@ func TestLandingPage_Tcp(t *testing.T) {
// start base with the listener and wait for it to be started
address, err := config.HTTPAddress(s.URL)
assert.NoError(t, err)
- go b.SetupAndServeHTTP(address, nil, nil)
+ go basepkg.SetupAndServeHTTP(processCtx, &cfg, routers, address, nil, nil)
time.Sleep(time.Millisecond * 10)
// When hitting /, we should be redirected to /_matrix/static, which should contain the landing page
@@ -70,15 +74,17 @@ func TestLandingPage_UnixSocket(t *testing.T) {
})
assert.NoError(t, err)
- b, _, _ := testrig.Base(nil)
- defer b.Close()
+ processCtx := process.NewProcessContext()
+ routers := httputil.NewRouters()
+ cfg := config.Dendrite{}
+ cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
tempDir := t.TempDir()
socket := path.Join(tempDir, "socket")
// start base with the listener and wait for it to be started
address, err := config.UnixSocketAddress(socket, "755")
assert.NoError(t, err)
- go b.SetupAndServeHTTP(address, nil, nil)
+ go basepkg.SetupAndServeHTTP(processCtx, &cfg, routers, address, nil, nil)
time.Sleep(time.Millisecond * 100)
client := &http.Client{
diff --git a/setup/base/sanity_other.go b/setup/base/sanity_other.go
index 48fe6e1f..d35c2e87 100644
--- a/setup/base/sanity_other.go
+++ b/setup/base/sanity_other.go
@@ -3,6 +3,6 @@
package base
-func platformSanityChecks() {
+func PlatformSanityChecks() {
// Nothing to do yet.
}
diff --git a/setup/base/sanity_unix.go b/setup/base/sanity_unix.go
index c630d3f1..0403df1a 100644
--- a/setup/base/sanity_unix.go
+++ b/setup/base/sanity_unix.go
@@ -9,7 +9,7 @@ import (
"github.com/sirupsen/logrus"
)
-func platformSanityChecks() {
+func PlatformSanityChecks() {
// Dendrite needs a relatively high number of file descriptors in order
// to function properly, particularly when federating with lots of servers.
// If we run out of file descriptors, we might run into problems accessing
diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go
index 48683789..06a58d54 100644
--- a/setup/jetstream/nats.go
+++ b/setup/jetstream/nats.go
@@ -56,7 +56,9 @@ func (s *NATSInstance) Prepare(process *process.ProcessContext, cfg *config.JetS
if err != nil {
panic(err)
}
- s.SetLogger(NewLogAdapter(), opts.Debug, opts.Trace)
+ if !cfg.NoLog {
+ s.SetLogger(NewLogAdapter(), opts.Debug, opts.Trace)
+ }
go func() {
process.ComponentStarted()
s.Start()
diff --git a/setup/monolith.go b/setup/monolith.go
index 174eba68..51e40070 100644
--- a/setup/monolith.go
+++ b/setup/monolith.go
@@ -21,13 +21,16 @@ import (
"github.com/matrix-org/dendrite/federationapi"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/internal/transactions"
"github.com/matrix-org/dendrite/mediaapi"
"github.com/matrix-org/dendrite/relayapi"
relayAPI "github.com/matrix-org/dendrite/relayapi/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/syncapi"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
@@ -53,23 +56,31 @@ type Monolith struct {
}
// AddAllPublicRoutes attaches all public paths to the given router
-func (m *Monolith) AddAllPublicRoutes(base *base.BaseDendrite, caches *caching.Caches) {
+func (m *Monolith) AddAllPublicRoutes(
+ processCtx *process.ProcessContext,
+ cfg *config.Dendrite,
+ routers httputil.Routers,
+ cm sqlutil.Connections,
+ natsInstance *jetstream.NATSInstance,
+ caches *caching.Caches,
+ enableMetrics bool,
+) {
userDirectoryProvider := m.ExtUserDirectoryProvider
if userDirectoryProvider == nil {
userDirectoryProvider = m.UserAPI
}
clientapi.AddPublicRoutes(
- base, m.FedClient, m.RoomserverAPI, m.AppserviceAPI, transactions.New(),
+ processCtx, routers, cfg, natsInstance, m.FedClient, m.RoomserverAPI, m.AppserviceAPI, transactions.New(),
m.FederationAPI, m.UserAPI, userDirectoryProvider,
- m.ExtPublicRoomsProvider,
+ m.ExtPublicRoomsProvider, enableMetrics,
)
federationapi.AddPublicRoutes(
- base, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, nil,
+ processCtx, routers, cfg, natsInstance, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, nil, enableMetrics,
)
- mediaapi.AddPublicRoutes(base, m.UserAPI, m.Client)
- syncapi.AddPublicRoutes(base, m.UserAPI, m.RoomserverAPI, caches)
+ mediaapi.AddPublicRoutes(routers.Media, cm, cfg, m.UserAPI, m.Client)
+ syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, natsInstance, m.UserAPI, m.RoomserverAPI, caches, enableMetrics)
if m.RelayAPI != nil {
- relayapi.AddPublicRoutes(base, m.KeyRing, m.RelayAPI)
+ relayapi.AddPublicRoutes(routers, cfg, m.KeyRing, m.RelayAPI)
}
}
diff --git a/setup/mscs/msc2836/msc2836.go b/setup/mscs/msc2836/msc2836.go
index 7c1e0fc6..92248bc1 100644
--- a/setup/mscs/msc2836/msc2836.go
+++ b/setup/mscs/msc2836/msc2836.go
@@ -31,8 +31,9 @@ import (
fs "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/hooks"
"github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
+ "github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@@ -99,10 +100,10 @@ func toClientResponse(res *MSC2836EventRelationshipsResponse) *EventRelationship
// Enable this MSC
func Enable(
- base *base.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
+ cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
userAPI userapi.UserInternalAPI, keyRing gomatrixserverlib.JSONVerifier,
) error {
- db, err := NewDatabase(base.ConnectionManager, &base.Cfg.MSCs.Database)
+ db, err := NewDatabase(cm, &cfg.MSCs.Database)
if err != nil {
return fmt.Errorf("cannot enable MSC2836: %w", err)
}
@@ -125,14 +126,14 @@ func Enable(
}
})
- base.Routers.Client.Handle("/unstable/event_relationships",
+ routers.Client.Handle("/unstable/event_relationships",
httputil.MakeAuthAPI("eventRelationships", userAPI, eventRelationshipHandler(db, rsAPI, fsAPI)),
).Methods(http.MethodPost, http.MethodOptions)
- base.Routers.Federation.Handle("/unstable/event_relationships", httputil.MakeExternalAPI(
+ routers.Federation.Handle("/unstable/event_relationships", httputil.MakeExternalAPI(
"msc2836_event_relationships", func(req *http.Request) util.JSONResponse {
fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest(
- req, time.Now(), base.Cfg.Global.ServerName, base.Cfg.Global.IsLocalServerName, keyRing,
+ req, time.Now(), cfg.Global.ServerName, cfg.Global.IsLocalServerName, keyRing,
)
if fedReq == nil {
return errResp
diff --git a/setup/mscs/msc2836/msc2836_test.go b/setup/mscs/msc2836/msc2836_test.go
index 24e96f93..bfcabef6 100644
--- a/setup/mscs/msc2836/msc2836_test.go
+++ b/setup/mscs/msc2836/msc2836_test.go
@@ -15,13 +15,13 @@ import (
"time"
"github.com/gorilla/mux"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal/hooks"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/mscs/msc2836"
userapi "github.com/matrix-org/dendrite/userapi/api"
@@ -555,21 +555,18 @@ func injectEvents(t *testing.T, userAPI userapi.UserInternalAPI, rsAPI roomserve
cfg.Global.ServerName = "localhost"
cfg.MSCs.Database.ConnectionString = "file:msc2836_test.db"
cfg.MSCs.MSCs = []string{"msc2836"}
- cm := sqlutil.NewConnectionManager()
- base := &base.BaseDendrite{
- Cfg: cfg,
- Routers: httputil.NewRouters(),
- ConnectionManager: cm,
- }
- err := msc2836.Enable(base, rsAPI, nil, userAPI, nil)
+ processCtx := process.NewProcessContext()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ routers := httputil.NewRouters()
+ err := msc2836.Enable(cfg, cm, routers, rsAPI, nil, userAPI, nil)
if err != nil {
t.Fatalf("failed to enable MSC2836: %s", err)
}
for _, ev := range events {
hooks.Run(hooks.KindNewEventPersisted, ev)
}
- return base.Routers.Client
+ return routers.Client
}
type fledglingEvent struct {
diff --git a/setup/mscs/msc2946/msc2946.go b/setup/mscs/msc2946/msc2946.go
index b4b93ff3..d33897b9 100644
--- a/setup/mscs/msc2946/msc2946.go
+++ b/setup/mscs/msc2946/msc2946.go
@@ -33,7 +33,7 @@ import (
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
+ "github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@@ -54,17 +54,17 @@ type MSC2946ClientResponse struct {
// Enable this MSC
func Enable(
- base *base.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, userAPI userapi.UserInternalAPI,
+ cfg *config.Dendrite, routers httputil.Routers, rsAPI roomserver.RoomserverInternalAPI, userAPI userapi.UserInternalAPI,
fsAPI fs.FederationInternalAPI, keyRing gomatrixserverlib.JSONVerifier, cache caching.SpaceSummaryRoomsCache,
) error {
- clientAPI := httputil.MakeAuthAPI("spaces", userAPI, spacesHandler(rsAPI, fsAPI, cache, base.Cfg.Global.ServerName), httputil.WithAllowGuests())
- base.Routers.Client.Handle("/v1/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
- base.Routers.Client.Handle("/unstable/org.matrix.msc2946/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
+ clientAPI := httputil.MakeAuthAPI("spaces", userAPI, spacesHandler(rsAPI, fsAPI, cache, cfg.Global.ServerName), httputil.WithAllowGuests())
+ routers.Client.Handle("/v1/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
+ routers.Client.Handle("/unstable/org.matrix.msc2946/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
fedAPI := httputil.MakeExternalAPI(
"msc2946_fed_spaces", func(req *http.Request) util.JSONResponse {
fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest(
- req, time.Now(), base.Cfg.Global.ServerName, base.Cfg.Global.IsLocalServerName, keyRing,
+ req, time.Now(), cfg.Global.ServerName, cfg.Global.IsLocalServerName, keyRing,
)
if fedReq == nil {
return errResp
@@ -75,11 +75,11 @@ func Enable(
return util.ErrorResponse(err)
}
roomID := params["roomID"]
- return federatedSpacesHandler(req.Context(), fedReq, roomID, cache, rsAPI, fsAPI, base.Cfg.Global.ServerName)
+ return federatedSpacesHandler(req.Context(), fedReq, roomID, cache, rsAPI, fsAPI, cfg.Global.ServerName)
},
)
- base.Routers.Federation.Handle("/unstable/org.matrix.msc2946/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
- base.Routers.Federation.Handle("/v1/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
+ routers.Federation.Handle("/unstable/org.matrix.msc2946/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
+ routers.Federation.Handle("/v1/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
return nil
}
diff --git a/setup/mscs/mscs.go b/setup/mscs/mscs.go
index b58c800b..9cd5eed1 100644
--- a/setup/mscs/mscs.go
+++ b/setup/mscs/mscs.go
@@ -20,30 +20,32 @@ import (
"fmt"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup"
- "github.com/matrix-org/dendrite/setup/base"
+ "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/mscs/msc2836"
"github.com/matrix-org/dendrite/setup/mscs/msc2946"
"github.com/matrix-org/util"
)
// Enable MSCs - returns an error on unknown MSCs
-func Enable(base *base.BaseDendrite, monolith *setup.Monolith, caches *caching.Caches) error {
- for _, msc := range base.Cfg.MSCs.MSCs {
+func Enable(cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, caches *caching.Caches) error {
+ for _, msc := range cfg.MSCs.MSCs {
util.GetLogger(context.Background()).WithField("msc", msc).Info("Enabling MSC")
- if err := EnableMSC(base, monolith, msc, caches); err != nil {
+ if err := EnableMSC(cfg, cm, routers, monolith, msc, caches); err != nil {
return err
}
}
return nil
}
-func EnableMSC(base *base.BaseDendrite, monolith *setup.Monolith, msc string, caches *caching.Caches) error {
+func EnableMSC(cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, msc string, caches *caching.Caches) error {
switch msc {
case "msc2836":
- return msc2836.Enable(base, monolith.RoomserverAPI, monolith.FederationAPI, monolith.UserAPI, monolith.KeyRing)
+ return msc2836.Enable(cfg, cm, routers, monolith.RoomserverAPI, monolith.FederationAPI, monolith.UserAPI, monolith.KeyRing)
case "msc2946":
- return msc2946.Enable(base, monolith.RoomserverAPI, monolith.UserAPI, monolith.FederationAPI, monolith.KeyRing, caches)
+ return msc2946.Enable(cfg, routers, monolith.RoomserverAPI, monolith.UserAPI, monolith.FederationAPI, monolith.KeyRing, caches)
case "msc2444": // enabled inside federationapi
case "msc2753": // enabled inside clientapi
default:
diff --git a/syncapi/storage/storage_test.go b/syncapi/storage/storage_test.go
index 9f006490..e81a341f 100644
--- a/syncapi/storage/storage_test.go
+++ b/syncapi/storage/storage_test.go
@@ -22,7 +22,7 @@ var ctx = context.Background()
func MustCreateDatabase(t *testing.T, dbType test.DBType) (storage.Database, func()) {
connStr, close := test.PrepareDBConnectionString(t, dbType)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewSyncServerDatasource(context.Background(), cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
})
diff --git a/syncapi/syncapi.go b/syncapi/syncapi.go
index e0cc8462..9a27b954 100644
--- a/syncapi/syncapi.go
+++ b/syncapi/syncapi.go
@@ -18,12 +18,15 @@ import (
"context"
"github.com/matrix-org/dendrite/internal/fulltext"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
userapi "github.com/matrix-org/dendrite/userapi/api"
@@ -39,16 +42,19 @@ import (
// AddPublicRoutes sets up and registers HTTP handlers for the SyncAPI
// component.
func AddPublicRoutes(
- base *base.BaseDendrite,
+ processContext *process.ProcessContext,
+ routers httputil.Routers,
+ dendriteCfg *config.Dendrite,
+ cm sqlutil.Connections,
+ natsInstance *jetstream.NATSInstance,
userAPI userapi.SyncUserAPI,
rsAPI api.SyncRoomserverAPI,
caches caching.LazyLoadCache,
+ enableMetrics bool,
) {
- cfg := &base.Cfg.SyncAPI
+ js, natsClient := natsInstance.Prepare(processContext, &dendriteCfg.Global.JetStream)
- js, natsClient := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
-
- syncDB, err := storage.NewSyncServerDatasource(base.Context(), base.ConnectionManager, &cfg.Database)
+ syncDB, err := storage.NewSyncServerDatasource(processContext.Context(), cm, &dendriteCfg.SyncAPI.Database)
if err != nil {
logrus.WithError(err).Panicf("failed to connect to sync db")
}
@@ -62,32 +68,32 @@ func AddPublicRoutes(
}
var fts *fulltext.Search
- if cfg.Fulltext.Enabled {
- fts, err = fulltext.New(base.ProcessContext.Context(), cfg.Fulltext)
+ if dendriteCfg.SyncAPI.Fulltext.Enabled {
+ fts, err = fulltext.New(processContext.Context(), dendriteCfg.SyncAPI.Fulltext)
if err != nil {
logrus.WithError(err).Panicf("failed to create full text")
}
- base.ProcessContext.ComponentStarted()
+ processContext.ComponentStarted()
}
federationPresenceProducer := &producers.FederationAPIPresenceProducer{
- Topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
+ Topic: dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
JetStream: js,
}
presenceConsumer := consumers.NewPresenceConsumer(
- base.ProcessContext, cfg, js, natsClient, syncDB,
+ processContext, &dendriteCfg.SyncAPI, js, natsClient, syncDB,
notifier, streams.PresenceStreamProvider,
userAPI,
)
- requestPool := sync.NewRequestPool(syncDB, cfg, userAPI, rsAPI, streams, notifier, federationPresenceProducer, presenceConsumer, base.EnableMetrics)
+ requestPool := sync.NewRequestPool(syncDB, &dendriteCfg.SyncAPI, userAPI, rsAPI, streams, notifier, federationPresenceProducer, presenceConsumer, enableMetrics)
if err = presenceConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start presence consumer")
}
keyChangeConsumer := consumers.NewOutputKeyChangeEventConsumer(
- base.ProcessContext, cfg, cfg.Matrix.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
+ processContext, &dendriteCfg.SyncAPI, dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
js, rsAPI, syncDB, notifier,
streams.DeviceListStreamProvider,
)
@@ -96,7 +102,7 @@ func AddPublicRoutes(
}
roomConsumer := consumers.NewOutputRoomEventConsumer(
- base.ProcessContext, cfg, js, syncDB, notifier, streams.PDUStreamProvider,
+ processContext, &dendriteCfg.SyncAPI, js, syncDB, notifier, streams.PDUStreamProvider,
streams.InviteStreamProvider, rsAPI, fts,
)
if err = roomConsumer.Start(); err != nil {
@@ -104,7 +110,7 @@ func AddPublicRoutes(
}
clientConsumer := consumers.NewOutputClientDataConsumer(
- base.ProcessContext, cfg, js, natsClient, syncDB, notifier,
+ processContext, &dendriteCfg.SyncAPI, js, natsClient, syncDB, notifier,
streams.AccountDataStreamProvider, fts,
)
if err = clientConsumer.Start(); err != nil {
@@ -112,35 +118,35 @@ func AddPublicRoutes(
}
notificationConsumer := consumers.NewOutputNotificationDataConsumer(
- base.ProcessContext, cfg, js, syncDB, notifier, streams.NotificationDataStreamProvider,
+ processContext, &dendriteCfg.SyncAPI, js, syncDB, notifier, streams.NotificationDataStreamProvider,
)
if err = notificationConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start notification data consumer")
}
typingConsumer := consumers.NewOutputTypingEventConsumer(
- base.ProcessContext, cfg, js, eduCache, notifier, streams.TypingStreamProvider,
+ processContext, &dendriteCfg.SyncAPI, js, eduCache, notifier, streams.TypingStreamProvider,
)
if err = typingConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start typing consumer")
}
sendToDeviceConsumer := consumers.NewOutputSendToDeviceEventConsumer(
- base.ProcessContext, cfg, js, syncDB, userAPI, notifier, streams.SendToDeviceStreamProvider,
+ processContext, &dendriteCfg.SyncAPI, js, syncDB, userAPI, notifier, streams.SendToDeviceStreamProvider,
)
if err = sendToDeviceConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start send-to-device consumer")
}
receiptConsumer := consumers.NewOutputReceiptEventConsumer(
- base.ProcessContext, cfg, js, syncDB, notifier, streams.ReceiptStreamProvider,
+ processContext, &dendriteCfg.SyncAPI, js, syncDB, notifier, streams.ReceiptStreamProvider,
)
if err = receiptConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start receipts consumer")
}
routing.Setup(
- base.Routers.Client, requestPool, syncDB, userAPI,
- rsAPI, cfg, caches, fts,
+ routers.Client, requestPool, syncDB, userAPI,
+ rsAPI, &dendriteCfg.SyncAPI, caches, fts,
)
}
diff --git a/syncapi/syncapi_test.go b/syncapi/syncapi_test.go
index 13a07865..584782af 100644
--- a/syncapi/syncapi_test.go
+++ b/syncapi/syncapi_test.go
@@ -11,6 +11,9 @@ import (
"time"
"github.com/matrix-org/dendrite/internal/caching"
+ "github.com/matrix-org/dendrite/internal/httputil"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
"github.com/tidwall/gjson"
@@ -22,7 +25,6 @@ import (
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/roomserver/api"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/dendrite/test"
@@ -114,14 +116,17 @@ func testSyncAccessTokens(t *testing.T, dbType test.DBType) {
AccountType: userapi.AccountTypeUser,
}
- base, close := testrig.CreateBaseDendrite(t, dbType)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ natsInstance := jetstream.NATSInstance{}
defer close()
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
- msgs := toNATSMsgs(t, base, room.Events()...)
- AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches)
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
+ msgs := toNATSMsgs(t, cfg, room.Events()...)
+ AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, caching.DisableMetrics)
testrig.MustPublishMsgs(t, jsctx, msgs...)
testCases := []struct {
@@ -156,7 +161,7 @@ func testSyncAccessTokens(t *testing.T, dbType test.DBType) {
},
}
- syncUntil(t, base, alice.AccessToken, false, func(syncBody string) bool {
+ syncUntil(t, routers, alice.AccessToken, false, func(syncBody string) bool {
// wait for the last sent eventID to come down sync
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(event_id=="%s")`, room.ID, room.Events()[len(room.Events())-1].EventID())
return gjson.Get(syncBody, path).Exists()
@@ -164,7 +169,7 @@ func testSyncAccessTokens(t *testing.T, dbType test.DBType) {
for _, tc := range testCases {
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, tc.req)
+ routers.Client.ServeHTTP(w, tc.req)
if w.Code != tc.wantCode {
t.Fatalf("%s: got HTTP %d want %d", tc.name, w.Code, tc.wantCode)
}
@@ -207,26 +212,29 @@ func testSyncAPICreateRoomSyncEarly(t *testing.T, dbType test.DBType) {
AccountType: userapi.AccountTypeUser,
}
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
+ natsInstance := jetstream.NATSInstance{}
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// order is:
// m.room.create
// m.room.member
// m.room.power_levels
// m.room.join_rules
// m.room.history_visibility
- msgs := toNATSMsgs(t, base, room.Events()...)
+ msgs := toNATSMsgs(t, cfg, room.Events()...)
sinceTokens := make([]string, len(msgs))
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches)
+ AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, caching.DisableMetrics)
for i, msg := range msgs {
testrig.MustPublishMsgs(t, jsctx, msg)
time.Sleep(100 * time.Millisecond)
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"timeout": "0",
})))
@@ -256,7 +264,7 @@ func testSyncAPICreateRoomSyncEarly(t *testing.T, dbType test.DBType) {
t.Logf("waited for events to be consumed; syncing with %v", sinceTokens)
for i, since := range sinceTokens {
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"timeout": "0",
"since": since,
@@ -298,17 +306,20 @@ func testSyncAPIUpdatePresenceImmediately(t *testing.T, dbType test.DBType) {
AccountType: userapi.AccountTypeUser,
}
- base, close := testrig.CreateBaseDendrite(t, dbType)
- base.Cfg.Global.Presence.EnableOutbound = true
- base.Cfg.Global.Presence.EnableInbound = true
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ cfg.Global.Presence.EnableOutbound = true
+ cfg.Global.Presence.EnableInbound = true
defer close()
+ natsInstance := jetstream.NATSInstance{}
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches)
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
+ AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches, caching.DisableMetrics)
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"timeout": "0",
"set_presence": "online",
@@ -414,17 +425,20 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
userType = "real user"
}
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
+ natsInstance := jetstream.NATSInstance{}
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// Use the actual internal roomserver API
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
- AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches)
+ AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, caching.DisableMetrics)
for _, tc := range testCases {
testname := fmt.Sprintf("%s - %s", tc.historyVisibility, userType)
@@ -439,7 +453,7 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
if err := api.SendEvents(ctx, rsAPI, api.KindNew, eventsToSend, "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
- syncUntil(t, base, aliceDev.AccessToken, false,
+ syncUntil(t, routers, aliceDev.AccessToken, false,
func(syncBody string) bool {
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(content.body=="%s")`, room.ID, beforeJoinBody)
return gjson.Get(syncBody, path).Exists()
@@ -448,7 +462,7 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
// There is only one event, we expect only to be able to see this, if the room is world_readable
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
"access_token": bobDev.AccessToken,
"dir": "b",
"filter": `{"lazy_load_members":true}`, // check that lazy loading doesn't break history visibility
@@ -479,7 +493,7 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
if err := api.SendEvents(ctx, rsAPI, api.KindNew, eventsToSend, "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
- syncUntil(t, base, aliceDev.AccessToken, false,
+ syncUntil(t, routers, aliceDev.AccessToken, false,
func(syncBody string) bool {
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(content.body=="%s")`, room.ID, afterJoinBody)
return gjson.Get(syncBody, path).Exists()
@@ -488,7 +502,7 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
// Verify the messages after/before invite are visible or not
w = httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
"access_token": bobDev.AccessToken,
"dir": "b",
})))
@@ -714,18 +728,20 @@ func TestGetMembership(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
-
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
+ natsInstance := jetstream.NATSInstance{}
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// Use an actual roomserver for this
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
- AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches)
+ AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, caching.DisableMetrics)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
@@ -745,7 +761,7 @@ func TestGetMembership(t *testing.T) {
if tc.useSleep {
time.Sleep(time.Millisecond * 100)
} else {
- syncUntil(t, base, aliceDev.AccessToken, false, func(syncBody string) bool {
+ syncUntil(t, routers, aliceDev.AccessToken, false, func(syncBody string) bool {
// wait for the last sent eventID to come down sync
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(event_id=="%s")`, room.ID, room.Events()[len(room.Events())-1].EventID())
return gjson.Get(syncBody, path).Exists()
@@ -753,7 +769,7 @@ func TestGetMembership(t *testing.T) {
}
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, tc.request(t, room))
+ routers.Client.ServeHTTP(w, tc.request(t, room))
if w.Code != 200 && tc.wantOK {
t.Logf("%s", w.Body.String())
t.Fatalf("got HTTP %d want %d", w.Code, 200)
@@ -786,16 +802,19 @@ func testSendToDevice(t *testing.T, dbType test.DBType) {
AccountType: userapi.AccountTypeUser,
}
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ defer close()
+ natsInstance := jetstream.NATSInstance{}
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches)
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
+ AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches, caching.DisableMetrics)
producer := producers.SyncAPIProducer{
- TopicSendToDeviceEvent: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
+ TopicSendToDeviceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
JetStream: jsctx,
}
@@ -881,7 +900,7 @@ func testSendToDevice(t *testing.T, dbType test.DBType) {
}
}
- syncUntil(t, base, alice.AccessToken,
+ syncUntil(t, routers, alice.AccessToken,
len(tc.want) == 0,
func(body string) bool {
return gjson.Get(body, fmt.Sprintf(`to_device.events.#(content.dummy=="message %d")`, msgCounter)).Exists()
@@ -890,7 +909,7 @@ func testSendToDevice(t *testing.T, dbType test.DBType) {
// Execute a /sync request, recording the response
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"since": tc.since,
})))
@@ -1004,15 +1023,18 @@ func testContext(t *testing.T, dbType test.DBType) {
AccountType: userapi.AccountTypeUser,
}
- base, baseClose := testrig.CreateBaseDendrite(t, dbType)
- defer baseClose()
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ routers := httputil.NewRouters()
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
+ defer close()
// Use an actual roomserver for this
- caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
- rsAPI := roomserver.NewInternalAPI(base, caches)
+ natsInstance := jetstream.NATSInstance{}
+ rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
- AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, rsAPI, caches)
+ AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, rsAPI, caches, caching.DisableMetrics)
room := test.NewRoom(t, user)
@@ -1025,10 +1047,10 @@ func testContext(t *testing.T, dbType test.DBType) {
t.Fatalf("failed to send events: %v", err)
}
- jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
- defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
+ jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
+ defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
- syncUntil(t, base, alice.AccessToken, false, func(syncBody string) bool {
+ syncUntil(t, routers, alice.AccessToken, false, func(syncBody string) bool {
// wait for the last sent eventID to come down sync
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(event_id=="%s")`, room.ID, thirdMsg.EventID())
return gjson.Get(syncBody, path).Exists()
@@ -1055,7 +1077,7 @@ func testContext(t *testing.T, dbType test.DBType) {
params[k] = v
}
}
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", requestPath, test.WithQueryParams(params)))
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", requestPath, test.WithQueryParams(params)))
if tc.wantError && w.Code == 200 {
t.Fatalf("Expected an error, but got none")
@@ -1143,9 +1165,10 @@ func TestUpdateRelations(t *testing.T) {
room := test.NewRoom(t, alice)
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- base, shutdownBase := testrig.CreateBaseDendrite(t, dbType)
- t.Cleanup(shutdownBase)
- db, err := storage.NewSyncServerDatasource(base.Context(), base.ConnectionManager, &base.Cfg.SyncAPI.Database)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ t.Cleanup(close)
+ db, err := storage.NewSyncServerDatasource(processCtx.Context(), cm, &cfg.SyncAPI.Database)
if err != nil {
t.Fatal(err)
}
@@ -1167,10 +1190,11 @@ func TestUpdateRelations(t *testing.T) {
}
func syncUntil(t *testing.T,
- base *base.BaseDendrite, accessToken string,
+ routers httputil.Routers, accessToken string,
skip bool,
checkFunc func(syncBody string) bool,
) {
+ t.Helper()
if checkFunc == nil {
t.Fatalf("No checkFunc defined")
}
@@ -1184,7 +1208,7 @@ func syncUntil(t *testing.T,
go func() {
for {
w := httptest.NewRecorder()
- base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": accessToken,
"timeout": "1000",
})))
@@ -1202,14 +1226,14 @@ func syncUntil(t *testing.T,
}
}
-func toNATSMsgs(t *testing.T, base *base.BaseDendrite, input ...*gomatrixserverlib.HeaderedEvent) []*nats.Msg {
+func toNATSMsgs(t *testing.T, cfg *config.Dendrite, input ...*gomatrixserverlib.HeaderedEvent) []*nats.Msg {
result := make([]*nats.Msg, len(input))
for i, ev := range input {
var addsStateIDs []string
if ev.StateKey() != nil {
addsStateIDs = append(addsStateIDs, ev.EventID())
}
- result[i] = testrig.NewOutputEventMsg(t, base, ev.RoomID(), api.OutputEvent{
+ result[i] = testrig.NewOutputEventMsg(t, cfg, ev.RoomID(), api.OutputEvent{
Type: rsapi.OutputTypeNewRoomEvent,
NewRoomEvent: &rsapi.OutputNewRoomEvent{
Event: ev,
diff --git a/test/testrig/base.go b/test/testrig/base.go
index bb8fded2..fd85578e 100644
--- a/test/testrig/base.go
+++ b/test/testrig/base.go
@@ -19,13 +19,12 @@ import (
"path/filepath"
"testing"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/test"
- "github.com/nats-io/nats.go"
)
-func CreateBaseDendrite(t *testing.T, dbType test.DBType) (*base.BaseDendrite, func()) {
+func CreateConfig(t *testing.T, dbType test.DBType) (*config.Dendrite, *process.ProcessContext, func()) {
var cfg config.Dendrite
cfg.Defaults(config.DefaultOpts{
Generate: false,
@@ -33,6 +32,7 @@ func CreateBaseDendrite(t *testing.T, dbType test.DBType) (*base.BaseDendrite, f
})
cfg.Global.JetStream.InMemory = true
cfg.FederationAPI.KeyPerspectives = nil
+ ctx := process.NewProcessContext()
switch dbType {
case test.DBTypePostgres:
cfg.Global.Defaults(config.DefaultOpts{ // autogen a signing key
@@ -51,18 +51,18 @@ func CreateBaseDendrite(t *testing.T, dbType test.DBType) (*base.BaseDendrite, f
// use a distinct prefix else concurrent postgres/sqlite runs will clash since NATS will use
// the file system event with InMemory=true :(
cfg.Global.JetStream.TopicPrefix = fmt.Sprintf("Test_%d_", dbType)
- connStr, close := test.PrepareDBConnectionString(t, dbType)
+
+ connStr, closeDb := test.PrepareDBConnectionString(t, dbType)
cfg.Global.DatabaseOptions = config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
MaxOpenConnections: 10,
MaxIdleConnections: 2,
ConnMaxLifetimeSeconds: 60,
}
- base := base.NewBaseDendrite(&cfg, base.DisableMetrics)
- return base, func() {
- base.ShutdownDendrite()
- base.WaitForShutdown()
- close()
+ return &cfg, ctx, func() {
+ ctx.ShutdownDendrite()
+ ctx.WaitForShutdown()
+ closeDb()
}
case test.DBTypeSQLite:
cfg.Defaults(config.DefaultOpts{
@@ -86,30 +86,13 @@ func CreateBaseDendrite(t *testing.T, dbType test.DBType) (*base.BaseDendrite, f
cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "userapi.db"))
cfg.RelayAPI.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "relayapi.db"))
- base := base.NewBaseDendrite(&cfg, base.DisableMetrics)
- return base, func() {
- base.ShutdownDendrite()
- base.WaitForShutdown()
+ return &cfg, ctx, func() {
+ ctx.ShutdownDendrite()
+ ctx.WaitForShutdown()
t.Cleanup(func() {}) // removes t.TempDir, where all database files are created
}
default:
t.Fatalf("unknown db type: %v", dbType)
}
- return nil, nil
-}
-
-func Base(cfg *config.Dendrite) (*base.BaseDendrite, nats.JetStreamContext, *nats.Conn) {
- if cfg == nil {
- cfg = &config.Dendrite{}
- cfg.Defaults(config.DefaultOpts{
- Generate: true,
- SingleDatabase: false,
- })
- }
- cfg.Global.JetStream.InMemory = true
- cfg.SyncAPI.Fulltext.InMemory = true
- cfg.FederationAPI.KeyPerspectives = nil
- base := base.NewBaseDendrite(cfg, base.DisableMetrics)
- js, jc := base.NATS.Prepare(base.ProcessContext, &cfg.Global.JetStream)
- return base, js, jc
+ return &config.Dendrite{}, nil, func() {}
}
diff --git a/test/testrig/jetstream.go b/test/testrig/jetstream.go
index b880eea4..5f15cfb3 100644
--- a/test/testrig/jetstream.go
+++ b/test/testrig/jetstream.go
@@ -4,10 +4,10 @@ import (
"encoding/json"
"testing"
+ "github.com/matrix-org/dendrite/setup/config"
"github.com/nats-io/nats.go"
"github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
)
@@ -20,9 +20,9 @@ func MustPublishMsgs(t *testing.T, jsctx nats.JetStreamContext, msgs ...*nats.Ms
}
}
-func NewOutputEventMsg(t *testing.T, base *base.BaseDendrite, roomID string, update api.OutputEvent) *nats.Msg {
+func NewOutputEventMsg(t *testing.T, cfg *config.Dendrite, roomID string, update api.OutputEvent) *nats.Msg {
t.Helper()
- msg := nats.NewMsg(base.Cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent))
+ msg := nats.NewMsg(cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent))
msg.Header.Set(jetstream.RoomEventType, string(update.Type))
msg.Header.Set(jetstream.RoomID, roomID)
var err error
diff --git a/userapi/consumers/roomserver_test.go b/userapi/consumers/roomserver_test.go
index a54706db..4827ad47 100644
--- a/userapi/consumers/roomserver_test.go
+++ b/userapi/consumers/roomserver_test.go
@@ -21,7 +21,7 @@ import (
func mustCreateDatabase(t *testing.T, dbType test.DBType) (storage.UserDatabase, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewUserDatabase(context.Background(), cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, "", 4, 0, 0, "")
diff --git a/userapi/internal/device_list_update_test.go b/userapi/internal/device_list_update_test.go
index 6ccc4b5f..c0965a2c 100644
--- a/userapi/internal/device_list_update_test.go
+++ b/userapi/internal/device_list_update_test.go
@@ -364,7 +364,7 @@ func mustCreateKeyserverDB(t *testing.T, dbType test.DBType) (storage.KeyDatabas
t.Helper()
connStr, clearDB := test.PrepareDBConnectionString(t, dbType)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewKeyDatabase(cm, &config.DatabaseOptions{ConnectionString: config.DataSource(connStr)})
if err != nil {
t.Fatal(err)
diff --git a/userapi/internal/key_api_test.go b/userapi/internal/key_api_test.go
index ec315c73..de2a6d2c 100644
--- a/userapi/internal/key_api_test.go
+++ b/userapi/internal/key_api_test.go
@@ -16,7 +16,7 @@ import (
func mustCreateDatabase(t *testing.T, dbType test.DBType) (storage.KeyDatabase, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewKeyDatabase(cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
})
diff --git a/userapi/storage/storage_test.go b/userapi/storage/storage_test.go
index c6369ef9..cf7c5144 100644
--- a/userapi/storage/storage_test.go
+++ b/userapi/storage/storage_test.go
@@ -35,7 +35,7 @@ var (
func mustCreateUserDatabase(t *testing.T, dbType test.DBType) (storage.UserDatabase, func()) {
connStr, close := test.PrepareDBConnectionString(t, dbType)
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewUserDatabase(context.Background(), cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, "localhost", bcrypt.MinCost, openIDLifetimeMS, loginTokenLifetime, "_server")
@@ -576,8 +576,9 @@ func Test_Notification(t *testing.T) {
}
func mustCreateKeyDatabase(t *testing.T, dbType test.DBType) (storage.KeyDatabase, func()) {
- base, close := testrig.CreateBaseDendrite(t, dbType)
- db, err := storage.NewKeyDatabase(base.ConnectionManager, &base.Cfg.KeyServer.Database)
+ cfg, processCtx, close := testrig.CreateConfig(t, dbType)
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ db, err := storage.NewKeyDatabase(cm, &cfg.KeyServer.Database)
if err != nil {
t.Fatalf("failed to create new database: %v", err)
}
diff --git a/userapi/userapi.go b/userapi/userapi.go
index 3ada8020..6dcbc121 100644
--- a/userapi/userapi.go
+++ b/userapi/userapi.go
@@ -19,10 +19,12 @@ import (
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/pushgateway"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
+ "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/process"
"github.com/sirupsen/logrus"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
- "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/dendrite/userapi/consumers"
@@ -35,31 +37,33 @@ import (
// NewInternalAPI returns a concrete implementation of the internal API. Callers
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
func NewInternalAPI(
- base *base.BaseDendrite,
+ processContext *process.ProcessContext,
+ dendriteCfg *config.Dendrite,
+ cm sqlutil.Connections,
+ natsInstance *jetstream.NATSInstance,
rsAPI rsapi.UserRoomserverAPI,
fedClient fedsenderapi.KeyserverFederationAPI,
) *internal.UserInternalAPI {
- cfg := &base.Cfg.UserAPI
- js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
- appServices := base.Cfg.Derived.ApplicationServices
+ js, _ := natsInstance.Prepare(processContext, &dendriteCfg.Global.JetStream)
+ appServices := dendriteCfg.Derived.ApplicationServices
- pgClient := pushgateway.NewHTTPClient(cfg.PushGatewayDisableTLSValidation)
+ pgClient := pushgateway.NewHTTPClient(dendriteCfg.UserAPI.PushGatewayDisableTLSValidation)
db, err := storage.NewUserDatabase(
- base.ProcessContext.Context(),
- base.ConnectionManager,
- &cfg.AccountDatabase,
- cfg.Matrix.ServerName,
- cfg.BCryptCost,
- cfg.OpenIDTokenLifetimeMS,
+ processContext.Context(),
+ cm,
+ &dendriteCfg.UserAPI.AccountDatabase,
+ dendriteCfg.Global.ServerName,
+ dendriteCfg.UserAPI.BCryptCost,
+ dendriteCfg.UserAPI.OpenIDTokenLifetimeMS,
api.DefaultLoginTokenLifetime,
- cfg.Matrix.ServerNotices.LocalPart,
+ dendriteCfg.UserAPI.Matrix.ServerNotices.LocalPart,
)
if err != nil {
logrus.WithError(err).Panicf("failed to connect to accounts db")
}
- keyDB, err := storage.NewKeyDatabase(base.ConnectionManager, &base.Cfg.KeyServer.Database)
+ keyDB, err := storage.NewKeyDatabase(cm, &dendriteCfg.KeyServer.Database)
if err != nil {
logrus.WithError(err).Panicf("failed to connect to key db")
}
@@ -70,11 +74,11 @@ func NewInternalAPI(
// it's handled by clientapi, and hence uses its topic. When user
// API handles it for all account data, we can remove it from
// here.
- cfg.Matrix.JetStream.Prefixed(jetstream.OutputClientData),
- cfg.Matrix.JetStream.Prefixed(jetstream.OutputNotificationData),
+ dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputClientData),
+ dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputNotificationData),
)
keyChangeProducer := &producers.KeyChange{
- Topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
+ Topic: dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
JetStream: js,
DB: keyDB,
}
@@ -84,15 +88,15 @@ func NewInternalAPI(
KeyDatabase: keyDB,
SyncProducer: syncProducer,
KeyChangeProducer: keyChangeProducer,
- Config: cfg,
+ Config: &dendriteCfg.UserAPI,
AppServices: appServices,
RSAPI: rsAPI,
- DisableTLSValidation: cfg.PushGatewayDisableTLSValidation,
+ DisableTLSValidation: dendriteCfg.UserAPI.PushGatewayDisableTLSValidation,
PgClient: pgClient,
FedClient: fedClient,
}
- updater := internal.NewDeviceListUpdater(base.ProcessContext, keyDB, userAPI, keyChangeProducer, fedClient, 8, rsAPI, cfg.Matrix.ServerName) // 8 workers TODO: configurable
+ updater := internal.NewDeviceListUpdater(processContext, keyDB, userAPI, keyChangeProducer, fedClient, 8, rsAPI, dendriteCfg.Global.ServerName) // 8 workers TODO: configurable
userAPI.Updater = updater
// Remove users which we don't share a room with anymore
if err := updater.CleanUp(); err != nil {
@@ -106,28 +110,28 @@ func NewInternalAPI(
}()
dlConsumer := consumers.NewDeviceListUpdateConsumer(
- base.ProcessContext, cfg, js, updater,
+ processContext, &dendriteCfg.UserAPI, js, updater,
)
if err := dlConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start device list consumer")
}
sigConsumer := consumers.NewSigningKeyUpdateConsumer(
- base.ProcessContext, cfg, js, userAPI,
+ processContext, &dendriteCfg.UserAPI, js, userAPI,
)
if err := sigConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start signing key consumer")
}
receiptConsumer := consumers.NewOutputReceiptEventConsumer(
- base.ProcessContext, cfg, js, db, syncProducer, pgClient,
+ processContext, &dendriteCfg.UserAPI, js, db, syncProducer, pgClient,
)
if err := receiptConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start user API receipt consumer")
}
eventConsumer := consumers.NewOutputRoomEventConsumer(
- base.ProcessContext, cfg, js, db, pgClient, rsAPI, syncProducer,
+ processContext, &dendriteCfg.UserAPI, js, db, pgClient, rsAPI, syncProducer,
)
if err := eventConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start user API streamed event consumer")
@@ -136,15 +140,15 @@ func NewInternalAPI(
var cleanOldNotifs func()
cleanOldNotifs = func() {
logrus.Infof("Cleaning old notifications")
- if err := db.DeleteOldNotifications(base.Context()); err != nil {
+ if err := db.DeleteOldNotifications(processContext.Context()); err != nil {
logrus.WithError(err).Error("Failed to clean old notifications")
}
time.AfterFunc(time.Hour, cleanOldNotifs)
}
time.AfterFunc(time.Minute, cleanOldNotifs)
- if base.Cfg.Global.ReportStats.Enabled {
- go util.StartPhoneHomeCollector(time.Now(), base.Cfg, db)
+ if dendriteCfg.Global.ReportStats.Enabled {
+ go util.StartPhoneHomeCollector(time.Now(), dendriteCfg, db)
}
return userAPI
diff --git a/userapi/userapi_test.go b/userapi/userapi_test.go
index c2d4e5a2..03e65635 100644
--- a/userapi/userapi_test.go
+++ b/userapi/userapi_test.go
@@ -68,34 +68,25 @@ func MustMakeInternalAPI(t *testing.T, opts apiTestOpts, dbType test.DBType, pub
if opts.loginTokenLifetime == 0 {
opts.loginTokenLifetime = api.DefaultLoginTokenLifetime * time.Millisecond
}
- base, baseclose := testrig.CreateBaseDendrite(t, dbType)
- connStr, close := test.PrepareDBConnectionString(t, dbType)
+ cfg, ctx, close := testrig.CreateConfig(t, dbType)
sName := serverName
if opts.serverName != "" {
sName = gomatrixserverlib.ServerName(opts.serverName)
}
- cm := sqlutil.NewConnectionManager()
- ctx := context.Background()
- accountDB, err := storage.NewUserDatabase(ctx, cm, &config.DatabaseOptions{
- ConnectionString: config.DataSource(connStr),
- }, sName, bcrypt.MinCost, config.DefaultOpenIDTokenLifetimeMS, opts.loginTokenLifetime, "")
+ cm := sqlutil.NewConnectionManager(ctx, cfg.Global.DatabaseOptions)
+
+ accountDB, err := storage.NewUserDatabase(ctx.Context(), cm, &cfg.UserAPI.AccountDatabase, sName, bcrypt.MinCost, config.DefaultOpenIDTokenLifetimeMS, opts.loginTokenLifetime, "")
if err != nil {
t.Fatalf("failed to create account DB: %s", err)
}
- keyDB, err := storage.NewKeyDatabase(base.ConnectionManager, &config.DatabaseOptions{
- ConnectionString: config.DataSource(connStr),
- })
+ keyDB, err := storage.NewKeyDatabase(cm, &cfg.KeyServer.Database)
if err != nil {
t.Fatalf("failed to create key DB: %s", err)
}
- cfg := &config.UserAPI{
- Matrix: &config.Global{
- SigningIdentity: gomatrixserverlib.SigningIdentity{
- ServerName: sName,
- },
- },
+ cfg.Global.SigningIdentity = gomatrixserverlib.SigningIdentity{
+ ServerName: sName,
}
if publisher == nil {
@@ -107,12 +98,11 @@ func MustMakeInternalAPI(t *testing.T, opts apiTestOpts, dbType test.DBType, pub
return &internal.UserInternalAPI{
DB: accountDB,
KeyDatabase: keyDB,
- Config: cfg,
+ Config: &cfg.UserAPI,
SyncProducer: syncProducer,
KeyChangeProducer: keyChangeProducer,
}, accountDB, func() {
close()
- baseclose()
}
}
diff --git a/userapi/util/notify_test.go b/userapi/util/notify_test.go
index c899e3a7..69461ddd 100644
--- a/userapi/util/notify_test.go
+++ b/userapi/util/notify_test.go
@@ -77,7 +77,7 @@ func TestNotifyUserCountsAsync(t *testing.T) {
// Create DB and Dendrite base
connStr, close := test.PrepareDBConnectionString(t, dbType)
defer close()
- cm := sqlutil.NewConnectionManager()
+ cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
db, err := storage.NewUserDatabase(ctx, cm, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, "test", bcrypt.MinCost, 0, 0, "")
diff --git a/userapi/util/phonehomestats_test.go b/userapi/util/phonehomestats_test.go
index 4e931a1b..191a35c0 100644
--- a/userapi/util/phonehomestats_test.go
+++ b/userapi/util/phonehomestats_test.go
@@ -7,10 +7,10 @@ import (
"testing"
"time"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"golang.org/x/crypto/bcrypt"
"github.com/matrix-org/dendrite/internal"
- "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/dendrite/userapi/storage"
@@ -18,12 +18,10 @@ import (
func TestCollect(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
- b, _, _ := testrig.Base(nil)
- connStr, closeDB := test.PrepareDBConnectionString(t, dbType)
+ cfg, processCtx, closeDB := testrig.CreateConfig(t, dbType)
defer closeDB()
- db, err := storage.NewUserDatabase(b.Context(), b.ConnectionManager, &config.DatabaseOptions{
- ConnectionString: config.DataSource(connStr),
- }, "localhost", bcrypt.MinCost, 1000, 1000, "")
+ cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
+ db, err := storage.NewUserDatabase(processCtx.Context(), cm, &cfg.UserAPI.AccountDatabase, "localhost", bcrypt.MinCost, 1000, 1000, "")
if err != nil {
t.Error(err)
}
@@ -62,12 +60,12 @@ func TestCollect(t *testing.T) {
}))
defer srv.Close()
- b.Cfg.Global.ReportStats.Endpoint = srv.URL
+ cfg.Global.ReportStats.Endpoint = srv.URL
stats := phoneHomeStats{
prevData: timestampToRUUsage{},
serverName: "localhost",
startTime: time.Now(),
- cfg: b.Cfg,
+ cfg: cfg,
db: db,
isMonolith: false,
client: &http.Client{Timeout: time.Second},