mirror of
https://github.com/1f349/dendrite.git
synced 2024-11-09 22:42:58 +00:00
7863a405a5
Use `IsBlacklistedOrBackingOff` from the federation API to check if we should fetch devices. To reduce back pressure, we now only queue retrying servers if there's space in the channel.
163 lines
5.9 KiB
Go
163 lines
5.9 KiB
Go
// Copyright 2020 The Matrix.org Foundation C.I.C.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package userapi
|
|
|
|
import (
|
|
"time"
|
|
|
|
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
|
|
"github.com/matrix-org/dendrite/federationapi/statistics"
|
|
"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/matrix-org/gomatrixserverlib/spec"
|
|
"github.com/sirupsen/logrus"
|
|
|
|
rsapi "github.com/matrix-org/dendrite/roomserver/api"
|
|
"github.com/matrix-org/dendrite/setup/jetstream"
|
|
"github.com/matrix-org/dendrite/userapi/api"
|
|
"github.com/matrix-org/dendrite/userapi/consumers"
|
|
"github.com/matrix-org/dendrite/userapi/internal"
|
|
"github.com/matrix-org/dendrite/userapi/producers"
|
|
"github.com/matrix-org/dendrite/userapi/storage"
|
|
"github.com/matrix-org/dendrite/userapi/util"
|
|
)
|
|
|
|
// 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.
|
|
//
|
|
// Creating a new instance of the user API requires a roomserver API with a federation API set
|
|
// using its `SetFederationAPI` method, other you may get nil-dereference errors.
|
|
func NewInternalAPI(
|
|
processContext *process.ProcessContext,
|
|
dendriteCfg *config.Dendrite,
|
|
cm *sqlutil.Connections,
|
|
natsInstance *jetstream.NATSInstance,
|
|
rsAPI rsapi.UserRoomserverAPI,
|
|
fedClient fedsenderapi.KeyserverFederationAPI,
|
|
enableMetrics bool,
|
|
blacklistedOrBackingOffFn func(s spec.ServerName) (*statistics.ServerStatistics, error),
|
|
) *internal.UserInternalAPI {
|
|
js, _ := natsInstance.Prepare(processContext, &dendriteCfg.Global.JetStream)
|
|
appServices := dendriteCfg.Derived.ApplicationServices
|
|
|
|
pgClient := pushgateway.NewHTTPClient(dendriteCfg.UserAPI.PushGatewayDisableTLSValidation)
|
|
|
|
db, err := storage.NewUserDatabase(
|
|
processContext.Context(),
|
|
cm,
|
|
&dendriteCfg.UserAPI.AccountDatabase,
|
|
dendriteCfg.Global.ServerName,
|
|
dendriteCfg.UserAPI.BCryptCost,
|
|
dendriteCfg.UserAPI.OpenIDTokenLifetimeMS,
|
|
api.DefaultLoginTokenLifetime,
|
|
dendriteCfg.UserAPI.Matrix.ServerNotices.LocalPart,
|
|
)
|
|
if err != nil {
|
|
logrus.WithError(err).Panicf("failed to connect to accounts db")
|
|
}
|
|
|
|
keyDB, err := storage.NewKeyDatabase(cm, &dendriteCfg.KeyServer.Database)
|
|
if err != nil {
|
|
logrus.WithError(err).Panicf("failed to connect to key db")
|
|
}
|
|
|
|
syncProducer := producers.NewSyncAPI(
|
|
db, js,
|
|
// TODO: user API should handle syncs for account data. Right now,
|
|
// 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.
|
|
dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputClientData),
|
|
dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputNotificationData),
|
|
)
|
|
keyChangeProducer := &producers.KeyChange{
|
|
Topic: dendriteCfg.Global.JetStream.Prefixed(jetstream.OutputKeyChangeEvent),
|
|
JetStream: js,
|
|
DB: keyDB,
|
|
}
|
|
|
|
userAPI := &internal.UserInternalAPI{
|
|
DB: db,
|
|
KeyDatabase: keyDB,
|
|
SyncProducer: syncProducer,
|
|
KeyChangeProducer: keyChangeProducer,
|
|
Config: &dendriteCfg.UserAPI,
|
|
AppServices: appServices,
|
|
RSAPI: rsAPI,
|
|
DisableTLSValidation: dendriteCfg.UserAPI.PushGatewayDisableTLSValidation,
|
|
PgClient: pgClient,
|
|
FedClient: fedClient,
|
|
}
|
|
|
|
updater := internal.NewDeviceListUpdater(processContext, keyDB, userAPI, keyChangeProducer, fedClient, dendriteCfg.UserAPI.WorkerCount, rsAPI, dendriteCfg.Global.ServerName, enableMetrics, blacklistedOrBackingOffFn)
|
|
userAPI.Updater = updater
|
|
// Remove users which we don't share a room with anymore
|
|
if err := updater.CleanUp(); err != nil {
|
|
logrus.WithError(err).Error("failed to cleanup stale device lists")
|
|
}
|
|
|
|
go func() {
|
|
if err := updater.Start(); err != nil {
|
|
logrus.WithError(err).Panicf("failed to start device list updater")
|
|
}
|
|
}()
|
|
|
|
dlConsumer := consumers.NewDeviceListUpdateConsumer(
|
|
processContext, &dendriteCfg.UserAPI, js, updater,
|
|
)
|
|
if err := dlConsumer.Start(); err != nil {
|
|
logrus.WithError(err).Panic("failed to start device list consumer")
|
|
}
|
|
|
|
sigConsumer := consumers.NewSigningKeyUpdateConsumer(
|
|
processContext, &dendriteCfg.UserAPI, js, userAPI,
|
|
)
|
|
if err := sigConsumer.Start(); err != nil {
|
|
logrus.WithError(err).Panic("failed to start signing key consumer")
|
|
}
|
|
|
|
receiptConsumer := consumers.NewOutputReceiptEventConsumer(
|
|
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(
|
|
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")
|
|
}
|
|
|
|
var cleanOldNotifs func()
|
|
cleanOldNotifs = func() {
|
|
logrus.Infof("Cleaning old notifications")
|
|
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 dendriteCfg.Global.ReportStats.Enabled {
|
|
go util.StartPhoneHomeCollector(time.Now(), dendriteCfg, db)
|
|
}
|
|
|
|
return userAPI
|
|
}
|