mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-16 18:43:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Remove BaseDendrite from federation API

Browse source
This commit is contained in:
Till Faelligen 2023-03-20 13:53:00 +01:00
parent 1fd3b2894f
commit 16fb22cad5
No known key found for this signature in database
GPG key ID: ACCDC9606D472758
15 changed files with 54 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
build/dendritejs-pinecone/main.go
View file

@ -196,7 +196,7 @@ func startup() {
)
rsAPI.SetAppserviceAPI(asQuery)
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.EnableMetrics)
fedSenderAPI := federationapi.NewInternalAPI(base, &natsInstance, federation, rsAPI, caches, keyRing, true)
fedSenderAPI := federationapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true)
rsAPI.SetFederationAPI(fedSenderAPI, keyRing)
monolith := setup.Monolith{

2
build/gobind-yggdrasil/monolith.go
View file

@ -165,7 +165,7 @@ func (m *DendriteMonolith) Start() {
rsAPI := roomserver.NewInternalAPI(base, &natsInstance, caches)
fsAPI := federationapi.NewInternalAPI(
base, &natsInstance, federation, rsAPI, caches, keyRing, true,
base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
userAPI := userapi.NewInternalAPI(base, &natsInstance, rsAPI, federation)

2
clientapi/admin_test.go
View file

@ -161,7 +161,7 @@ func TestPurgeRoom(t *testing.T) {
// this starts the JetStream consumers
syncapi.AddPublicRoutes(base, &natsInstance, userAPI, rsAPI, caches)
federationapi.NewInternalAPI(base, &natsInstance, fedClient, rsAPI, caches, nil, true)
federationapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, fedClient, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(nil, nil)
// Create the room

2
cmd/dendrite-demo-pinecone/monolith/monolith.go
View file

@ -139,7 +139,7 @@ func (p *P2PMonolith) SetupDendrite(cfg *config.Dendrite, port int, enableRelayi
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(p.BaseDendrite, &natsInstance, caches)
fsAPI := federationapi.NewInternalAPI(
p.BaseDendrite, &natsInstance, federation, rsAPI, caches, keyRing, true,
p.BaseDendrite.ProcessContext, p.BaseDendrite.Cfg, p.BaseDendrite.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
userAPI := userapi.NewInternalAPI(p.BaseDendrite, &natsInstance, rsAPI, federation)

2
cmd/dendrite-demo-yggdrasil/main.go
View file

@ -167,7 +167,7 @@ func main() {
asAPI := appservice.NewInternalAPI(base.ProcessContext, base.Cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
fsAPI := federationapi.NewInternalAPI(
base, &natsInstance, federation, rsAPI, caches, keyRing, true,
base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
rsAPI.SetFederationAPI(fsAPI, keyRing)

2
cmd/dendrite/main.go
View file

@ -78,7 +78,7 @@ func main() {
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(base, &natsInstance, caches)
fsAPI := federationapi.NewInternalAPI(
base, &natsInstance, federation, rsAPI, caches, nil, false,
base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, nil, false,
)
keyRing := fsAPI.KeyRing()

49
federationapi/federationapi.go
View file

@ -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,7 +43,9 @@ 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,
@ -48,10 +53,11 @@ func AddPublicRoutes(
rsAPI roomserverAPI.FederationRoomserverAPI,
fedAPI federationAPI.FederationInternalAPI,
servers federationAPI.ServersInRoomProvider,
enableMetrics bool,
) {
cfg := &base.Cfg.FederationAPI
mscCfg := &base.Cfg.MSCs
js, _ := natsInstance.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),
@ -76,17 +82,20 @@ 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,
@ -94,9 +103,9 @@ func NewInternalAPI(
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")
}
@ -110,51 +119,51 @@ func NewInternalAPI(
cfg.FederationMaxRetries+1,
cfg.P2PFederationRetriesUntilAssumedOffline+1)
js, nats := natsInstance.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")
@ -163,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)

2
federationapi/federationapi_keys_test.go
View file

@ -111,7 +111,7 @@ func TestMain(m *testing.M) {
// Finally, build the server key APIs.
sbase := base.NewBaseDendrite(cfg, base.DisableMetrics)
s.api = NewInternalAPI(sbase, &natsInstance, s.fedclient, nil, s.cache, nil, true)
s.api = NewInternalAPI(sbase.ProcessContext, sbase.Cfg, sbase.ConnectionManager, &natsInstance, s.fedclient, nil, s.cache, nil, true)
}
// Now that we have built our server key APIs, start the

4
federationapi/federationapi_test.go
View file

@ -215,7 +215,7 @@ func testFederationAPIJoinThenKeyUpdate(t *testing.T, dbType test.DBType) {
},
},
}
fsapi := federationapi.NewInternalAPI(base, &natsInstance, fc, rsapi, caches, nil, false)
fsapi := federationapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, fc, rsapi, caches, nil, false)
var resp api.PerformJoinResponse
fsapi.PerformJoin(context.Background(), &api.PerformJoinRequest{
@ -281,7 +281,7 @@ func TestRoomsV3URLEscapeDoNot404(t *testing.T) {
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(b, &natsInstance, nil, nil, keyRing, nil, &internal.FederationInternalAPI{}, nil)
federationapi.AddPublicRoutes(b.ProcessContext, b.Routers, b.Cfg, &natsInstance, nil, nil, keyRing, nil, &internal.FederationInternalAPI{}, nil, false)
baseURL, cancel := test.ListenAndServe(t, b.Routers.Federation, true)
defer cancel()
serverName := gomatrixserverlib.ServerName(strings.TrimPrefix(baseURL, "https://"))

4
federationapi/routing/profile_test.go
View file

@ -58,13 +58,13 @@ func TestHandleQueryProfile(t *testing.T) {
fedClient := fakeFedClient{}
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
fedapi := fedAPI.NewInternalAPI(base, &natsInstance, &fedClient, nil, nil, keyRing, true)
fedapi := fedAPI.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &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(base.Routers, base.Cfg, nil, r, keyRing, &fedClient, &userapi, &base.Cfg.MSCs, nil, nil, base.EnableMetrics)
handler := fedMux.Get(routing.QueryProfileRouteName).GetHandler().ServeHTTP
_, sk, _ := ed25519.GenerateKey(nil)

4
federationapi/routing/query_test.go
View file

@ -58,13 +58,13 @@ func TestHandleQueryDirectory(t *testing.T) {
fedClient := fakeFedClient{}
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
fedapi := fedAPI.NewInternalAPI(base, &natsInstance, &fedClient, nil, nil, keyRing, true)
fedapi := fedAPI.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &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(base.Routers, base.Cfg, nil, r, keyRing, &fedClient, &userapi, &base.Cfg.MSCs, nil, nil, base.EnableMetrics)
handler := fedMux.Get(routing.QueryDirectoryRouteName).GetHandler().ServeHTTP
_, sk, _ := ed25519.GenerateKey(nil)

16
federationapi/routing/routing.go
View file

@ -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,
)

4
federationapi/routing/send_test.go
View file

@ -53,14 +53,14 @@ func TestHandleSend(t *testing.T) {
base.Routers.Federation = fedMux
base.Cfg.FederationAPI.Matrix.SigningIdentity.ServerName = testOrigin
base.Cfg.FederationAPI.Matrix.Metrics.Enabled = false
fedapi := fedAPI.NewInternalAPI(base, &natsInstance, nil, nil, nil, nil, true)
fedapi := fedAPI.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &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(base.Routers, base.Cfg, nil, r, keyRing, nil, nil, &base.Cfg.MSCs, nil, nil, base.EnableMetrics)
handler := fedMux.Get(routing.SendRouteName).GetHandler().ServeHTTP
_, sk, _ := ed25519.GenerateKey(nil)

2
roomserver/roomserver_test.go
View file

@ -245,7 +245,7 @@ func TestPurgeRoom(t *testing.T) {
// this starts the JetStream consumers
syncapi.AddPublicRoutes(base, &natsInstance, userAPI, rsAPI, caches)
federationapi.NewInternalAPI(base, &natsInstance, fedClient, rsAPI, caches, nil, true)
federationapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, fedClient, rsAPI, caches, nil, true)
rsAPI.SetFederationAPI(nil, nil)
// Create the room

2
setup/monolith.go
View file

@ -65,7 +65,7 @@ func (m *Monolith) AddAllPublicRoutes(base *base.BaseDendrite, natsInstance *jet
m.ExtPublicRoomsProvider, base.EnableMetrics,
)
federationapi.AddPublicRoutes(
base, natsInstance, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, nil,
base.ProcessContext, base.Routers, base.Cfg, natsInstance, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, nil, base.EnableMetrics,
)
mediaapi.AddPublicRoutes(base.Routers.Media, base.ConnectionManager, base.Cfg, m.UserAPI, m.Client)
syncapi.AddPublicRoutes(base, natsInstance, m.UserAPI, m.RoomserverAPI, caches)