mirror of
https://github.com/element-hq/dendrite.git
synced 2025-09-14 05:12:26 +03:00
Preparations for removing BaseDendrite (#3016)
Preparations to actually remove/replace `BaseDendrite`. Quite a few changes: - SyncAPI accepts an `fulltext.Indexer` interface (fulltext is removed from `BaseDendrite`) - Caches are removed from `BaseDendrite` - Introduces a `Router` struct (likely to change) - also fixes #2903 - Introduces a `sqlutil.ConnectionManager`, which should remove `base.DatabaseConnection` later on - probably more
This commit is contained in:
Till
GitHub
parent
d88f71ab71
commit
5579121c6f
85 changed files with 722 additions and 470 deletions
|
|
@ -10,6 +10,7 @@ import (
|
|||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal/caching"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/tidwall/gjson"
|
||||
|
|
@ -114,12 +115,13 @@ func testSyncAccessTokens(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)
|
||||
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}})
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches)
|
||||
testrig.MustPublishMsgs(t, jsctx, msgs...)
|
||||
|
||||
testCases := []struct {
|
||||
|
|
@ -162,7 +164,7 @@ func testSyncAccessTokens(t *testing.T, dbType test.DBType) {
|
|||
|
||||
for _, tc := range testCases {
|
||||
w := httptest.NewRecorder()
|
||||
base.PublicClientAPIMux.ServeHTTP(w, tc.req)
|
||||
base.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)
|
||||
}
|
||||
|
|
@ -218,12 +220,13 @@ func testSyncAPICreateRoomSyncEarly(t *testing.T, dbType test.DBType) {
|
|||
// m.room.history_visibility
|
||||
msgs := toNATSMsgs(t, base, room.Events()...)
|
||||
sinceTokens := make([]string, len(msgs))
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}})
|
||||
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)
|
||||
for i, msg := range msgs {
|
||||
testrig.MustPublishMsgs(t, jsctx, msg)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
w := httptest.NewRecorder()
|
||||
base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
"access_token": alice.AccessToken,
|
||||
"timeout": "0",
|
||||
})))
|
||||
|
|
@ -253,7 +256,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.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
base.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,
|
||||
|
|
@ -302,9 +305,10 @@ func testSyncAPIUpdatePresenceImmediately(t *testing.T, dbType test.DBType) {
|
|||
|
||||
jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
|
||||
defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{})
|
||||
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches)
|
||||
w := httptest.NewRecorder()
|
||||
base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
base.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",
|
||||
|
|
@ -417,10 +421,10 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
|
|||
defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
|
||||
|
||||
// Use the actual internal roomserver API
|
||||
rsAPI := roomserver.NewInternalAPI(base)
|
||||
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
|
||||
rsAPI := roomserver.NewInternalAPI(base, caches)
|
||||
rsAPI.SetFederationAPI(nil, nil)
|
||||
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI)
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches)
|
||||
|
||||
for _, tc := range testCases {
|
||||
testname := fmt.Sprintf("%s - %s", tc.historyVisibility, userType)
|
||||
|
|
@ -444,7 +448,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.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
|
||||
base.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
|
||||
|
|
@ -484,7 +488,7 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
|
|||
|
||||
// Verify the messages after/before invite are visible or not
|
||||
w = httptest.NewRecorder()
|
||||
base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
|
||||
base.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",
|
||||
})))
|
||||
|
|
@ -717,10 +721,11 @@ func TestGetMembership(t *testing.T) {
|
|||
defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
|
||||
|
||||
// Use an actual roomserver for this
|
||||
rsAPI := roomserver.NewInternalAPI(base)
|
||||
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
|
||||
rsAPI := roomserver.NewInternalAPI(base, caches)
|
||||
rsAPI.SetFederationAPI(nil, nil)
|
||||
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI)
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches)
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
|
|
@ -748,7 +753,7 @@ func TestGetMembership(t *testing.T) {
|
|||
}
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
base.PublicClientAPIMux.ServeHTTP(w, tc.request(t, room))
|
||||
base.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,8 +791,8 @@ func testSendToDevice(t *testing.T, dbType test.DBType) {
|
|||
|
||||
jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
|
||||
defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
|
||||
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{})
|
||||
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches)
|
||||
|
||||
producer := producers.SyncAPIProducer{
|
||||
TopicSendToDeviceEvent: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
|
||||
|
|
@ -885,7 +890,7 @@ func testSendToDevice(t *testing.T, dbType test.DBType) {
|
|||
|
||||
// Execute a /sync request, recording the response
|
||||
w := httptest.NewRecorder()
|
||||
base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
"access_token": alice.AccessToken,
|
||||
"since": tc.since,
|
||||
})))
|
||||
|
|
@ -1003,10 +1008,11 @@ func testContext(t *testing.T, dbType test.DBType) {
|
|||
defer baseClose()
|
||||
|
||||
// Use an actual roomserver for this
|
||||
rsAPI := roomserver.NewInternalAPI(base)
|
||||
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.DisableMetrics)
|
||||
rsAPI := roomserver.NewInternalAPI(base, caches)
|
||||
rsAPI.SetFederationAPI(nil, nil)
|
||||
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, rsAPI)
|
||||
AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, rsAPI, caches)
|
||||
|
||||
room := test.NewRoom(t, user)
|
||||
|
||||
|
|
@ -1049,7 +1055,7 @@ func testContext(t *testing.T, dbType test.DBType) {
|
|||
params[k] = v
|
||||
}
|
||||
}
|
||||
base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", requestPath, test.WithQueryParams(params)))
|
||||
base.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")
|
||||
|
|
@ -1139,7 +1145,7 @@ func TestUpdateRelations(t *testing.T) {
|
|||
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
|
||||
base, shutdownBase := testrig.CreateBaseDendrite(t, dbType)
|
||||
t.Cleanup(shutdownBase)
|
||||
db, err := storage.NewSyncServerDatasource(base, &base.Cfg.SyncAPI.Database)
|
||||
db, err := storage.NewSyncServerDatasource(base.Context(), base.ConnectionManager, &base.Cfg.SyncAPI.Database)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -1178,7 +1184,7 @@ func syncUntil(t *testing.T,
|
|||
go func() {
|
||||
for {
|
||||
w := httptest.NewRecorder()
|
||||
base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
base.Routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
|
||||
"access_token": accessToken,
|
||||
"timeout": "1000",
|
||||
})))
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue