sigb.us/dendrite
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases 1 Wiki Activity

Merge branch 'main' into s7evink/consent-tracking

Browse source
This commit is contained in:
kegsay 2022-05-09 16:04:36 +01:00 committed by GitHub
parent 3d3773d3d4 1a7f4c8aa9
commit c9409078ac
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
75 changed files with 897 additions and 639 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
appservice/api/query.go
View file

@ -26,6 +26,23 @@ import (
"github.com/matrix-org/gomatrixserverlib"
)
// AppServiceInternalAPI is used to query user and room alias data from application
// services
type AppServiceInternalAPI interface {
// Check whether a room alias exists within any application service namespaces
RoomAliasExists(
ctx context.Context,
req *RoomAliasExistsRequest,
resp *RoomAliasExistsResponse,
) error
// Check whether a user ID exists within any application service namespaces
UserIDExists(
ctx context.Context,
req *UserIDExistsRequest,
resp *UserIDExistsResponse,
) error
}
// RoomAliasExistsRequest is a request to an application service
// about whether a room alias exists
type RoomAliasExistsRequest struct {
@ -60,30 +77,13 @@ type UserIDExistsResponse struct {
UserIDExists bool `json:"exists"`
}
// AppServiceQueryAPI is used to query user and room alias data from application
// services
type AppServiceQueryAPI interface {
// Check whether a room alias exists within any application service namespaces
RoomAliasExists(
ctx context.Context,
req *RoomAliasExistsRequest,
resp *RoomAliasExistsResponse,
) error
// Check whether a user ID exists within any application service namespaces
UserIDExists(
ctx context.Context,
req *UserIDExistsRequest,
resp *UserIDExistsResponse,
) error
}
// RetrieveUserProfile is a wrapper that queries both the local database and
// application services for a given user's profile
// TODO: Remove this, it's called from federationapi and clientapi but is a pure function
func RetrieveUserProfile(
ctx context.Context,
userID string,
asAPI AppServiceQueryAPI,
asAPI AppServiceInternalAPI,
profileAPI userapi.ClientUserAPI,
) (*authtypes.Profile, error) {
localpart, _, err := gomatrixserverlib.SplitID('@', userID)

7
appservice/appservice.go
View file

@ -32,13 +32,12 @@ import (
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"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
)
// AddInternalRoutes registers HTTP handlers for internal API calls
func AddInternalRoutes(router *mux.Router, queryAPI appserviceAPI.AppServiceQueryAPI) {
func AddInternalRoutes(router *mux.Router, queryAPI appserviceAPI.AppServiceInternalAPI) {
inthttp.AddRoutes(queryAPI, router)
}
@ -48,14 +47,14 @@ func NewInternalAPI(
base *base.BaseDendrite,
userAPI userapi.AppserviceUserAPI,
rsAPI roomserverAPI.AppserviceRoomserverAPI,
) appserviceAPI.AppServiceQueryAPI {
) appserviceAPI.AppServiceInternalAPI {
client := gomatrixserverlib.NewClient(
gomatrixserverlib.WithTimeout(time.Second*30),
gomatrixserverlib.WithKeepAlives(false),
gomatrixserverlib.WithSkipVerify(base.Cfg.AppServiceAPI.DisableTLSValidation),
)
js, _ := jetstream.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
js, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
// Create a connection to the appservice postgres DB
appserviceDB, err := storage.NewDatabase(base, &base.Cfg.AppServiceAPI.Database)

2
appservice/inthttp/client.go
View file

@ -29,7 +29,7 @@ type httpAppServiceQueryAPI struct {
func NewAppserviceClient(
appserviceURL string,
httpClient *http.Client,
) (api.AppServiceQueryAPI, error) {
) (api.AppServiceInternalAPI, error) {
if httpClient == nil {
return nil, errors.New("NewRoomserverAliasAPIHTTP: httpClient is <nil>")
}

2
appservice/inthttp/server.go
View file

@ -11,7 +11,7 @@ import (
)
// AddRoutes adds the AppServiceQueryAPI handlers to the http.ServeMux.
func AddRoutes(a api.AppServiceQueryAPI, internalAPIMux *mux.Router) {
func AddRoutes(a api.AppServiceInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
AppServiceRoomAliasExistsPath,
httputil.MakeInternalAPI("appserviceRoomAliasExists", func(req *http.Request) util.JSONResponse {

4
clientapi/clientapi.go
View file

@ -34,7 +34,7 @@ func AddPublicRoutes(
base *base.BaseDendrite,
federation *gomatrixserverlib.FederationClient,
rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
transactionsCache *transactions.Cache,
fsAPI federationAPI.ClientFederationAPI,
userAPI userapi.ClientUserAPI,
@ -44,7 +44,7 @@ func AddPublicRoutes(
) {
cfg := &base.Cfg.ClientAPI
mscCfg := &base.Cfg.MSCs
js, natsClient := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
js, natsClient := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
syncProducer := &producers.SyncAPIProducer{
JetStream: js,

4
clientapi/routing/createroom.go
View file

@ -138,7 +138,7 @@ func CreateRoom(
req *http.Request, device *api.Device,
cfg *config.ClientAPI,
profileAPI api.ClientUserAPI, rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
) util.JSONResponse {
var r createRoomRequest
resErr := httputil.UnmarshalJSONRequest(req, &r)
@ -165,7 +165,7 @@ func createRoom(
r createRoomRequest, device *api.Device,
cfg *config.ClientAPI,
profileAPI api.ClientUserAPI, rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
evTime time.Time,
) util.JSONResponse {
// TODO (#267): Check room ID doesn't clash with an existing one, and we

16
clientapi/routing/membership.go
View file

@ -41,7 +41,7 @@ var errMissingUserID = errors.New("'user_id' must be supplied")
func SendBan(
req *http.Request, profileAPI userapi.ClientUserAPI, device *userapi.Device,
roomID string, cfg *config.ClientAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceQueryAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceInternalAPI,
) util.JSONResponse {
body, evTime, roomVer, reqErr := extractRequestData(req, roomID, rsAPI)
if reqErr != nil {
@ -84,7 +84,7 @@ func SendBan(
func sendMembership(ctx context.Context, profileAPI userapi.ClientUserAPI, device *userapi.Device,
roomID, membership, reason string, cfg *config.ClientAPI, targetUserID string, evTime time.Time,
roomVer gomatrixserverlib.RoomVersion,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceQueryAPI) util.JSONResponse {
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceInternalAPI) util.JSONResponse {
event, err := buildMembershipEvent(
ctx, targetUserID, reason, profileAPI, device, membership,
@ -127,7 +127,7 @@ func sendMembership(ctx context.Context, profileAPI userapi.ClientUserAPI, devic
func SendKick(
req *http.Request, profileAPI userapi.ClientUserAPI, device *userapi.Device,
roomID string, cfg *config.ClientAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceQueryAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceInternalAPI,
) util.JSONResponse {
body, evTime, roomVer, reqErr := extractRequestData(req, roomID, rsAPI)
if reqErr != nil {
@ -167,7 +167,7 @@ func SendKick(
func SendUnban(
req *http.Request, profileAPI userapi.ClientUserAPI, device *userapi.Device,
roomID string, cfg *config.ClientAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceQueryAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceInternalAPI,
) util.JSONResponse {
body, evTime, roomVer, reqErr := extractRequestData(req, roomID, rsAPI)
if reqErr != nil {
@ -202,7 +202,7 @@ func SendUnban(
func SendInvite(
req *http.Request, profileAPI userapi.ClientUserAPI, device *userapi.Device,
roomID string, cfg *config.ClientAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceQueryAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceInternalAPI,
) util.JSONResponse {
body, evTime, _, reqErr := extractRequestData(req, roomID, rsAPI)
if reqErr != nil {
@ -239,7 +239,7 @@ func sendInvite(
roomID, userID, reason string,
cfg *config.ClientAPI,
rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceQueryAPI, evTime time.Time,
asAPI appserviceAPI.AppServiceInternalAPI, evTime time.Time,
) (util.JSONResponse, error) {
event, err := buildMembershipEvent(
ctx, userID, reason, profileAPI, device, "invite",
@ -289,7 +289,7 @@ func buildMembershipEvent(
device *userapi.Device,
membership, roomID string, isDirect bool,
cfg *config.ClientAPI, evTime time.Time,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceQueryAPI,
rsAPI roomserverAPI.ClientRoomserverAPI, asAPI appserviceAPI.AppServiceInternalAPI,
) (*gomatrixserverlib.HeaderedEvent, error) {
profile, err := loadProfile(ctx, targetUserID, cfg, profileAPI, asAPI)
if err != nil {
@ -327,7 +327,7 @@ func loadProfile(
userID string,
cfg *config.ClientAPI,
profileAPI userapi.ClientUserAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
) (*authtypes.Profile, error) {
_, serverName, err := gomatrixserverlib.SplitID('@', userID)
if err != nil {

8
clientapi/routing/profile.go
View file

@ -37,7 +37,7 @@ import (
func GetProfile(
req *http.Request, profileAPI userapi.ClientUserAPI, cfg *config.ClientAPI,
userID string,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
federation *gomatrixserverlib.FederationClient,
) util.JSONResponse {
profile, err := getProfile(req.Context(), profileAPI, cfg, userID, asAPI, federation)
@ -65,7 +65,7 @@ func GetProfile(
// GetAvatarURL implements GET /profile/{userID}/avatar_url
func GetAvatarURL(
req *http.Request, profileAPI userapi.ClientUserAPI, cfg *config.ClientAPI,
userID string, asAPI appserviceAPI.AppServiceQueryAPI,
userID string, asAPI appserviceAPI.AppServiceInternalAPI,
federation *gomatrixserverlib.FederationClient,
) util.JSONResponse {
profile, err := getProfile(req.Context(), profileAPI, cfg, userID, asAPI, federation)
@ -194,7 +194,7 @@ func SetAvatarURL(
// GetDisplayName implements GET /profile/{userID}/displayname
func GetDisplayName(
req *http.Request, profileAPI userapi.ClientUserAPI, cfg *config.ClientAPI,
userID string, asAPI appserviceAPI.AppServiceQueryAPI,
userID string, asAPI appserviceAPI.AppServiceInternalAPI,
federation *gomatrixserverlib.FederationClient,
) util.JSONResponse {
profile, err := getProfile(req.Context(), profileAPI, cfg, userID, asAPI, federation)
@ -327,7 +327,7 @@ func SetDisplayName(
func getProfile(
ctx context.Context, profileAPI userapi.ClientUserAPI, cfg *config.ClientAPI,
userID string,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
federation *gomatrixserverlib.FederationClient,
) (*authtypes.Profile, error) {
localpart, domain, err := gomatrixserverlib.SplitID('@', userID)

2
clientapi/routing/routing.go
View file

@ -51,7 +51,7 @@ func Setup(
publicAPIMux, synapseAdminRouter, dendriteAdminRouter *mux.Router,
cfg *config.ClientAPI,
rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
userAPI userapi.ClientUserAPI,
userDirectoryProvider userapi.QuerySearchProfilesAPI,
federation *gomatrixserverlib.FederationClient,

2
clientapi/routing/server_notices.go
View file

@ -58,7 +58,7 @@ func SendServerNotice(
cfgClient *config.ClientAPI,
userAPI userapi.ClientUserAPI,
rsAPI api.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
device *userapi.Device,
senderDevice *userapi.Device,
txnID *string,

2
clientapi/routing/upgrade_room.go
View file

@ -42,7 +42,7 @@ func UpgradeRoom(
cfg *config.ClientAPI,
roomID string, profileAPI userapi.ClientUserAPI,
rsAPI roomserverAPI.ClientRoomserverAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
asAPI appserviceAPI.AppServiceInternalAPI,
) util.JSONResponse {
var r upgradeRoomRequest
if rErr := httputil.UnmarshalJSONRequest(req, &r); rErr != nil {

6
cmd/dendrite-monolith-server/main.go
View file

@ -89,6 +89,7 @@ func main() {
fsAPI := federationapi.NewInternalAPI(
base, federation, rsAPI, base.Caches, nil, false,
)
fsImplAPI := fsAPI
if base.UseHTTPAPIs {
federationapi.AddInternalRoutes(base.InternalAPIMux, fsAPI)
fsAPI = base.FederationAPIHTTPClient()
@ -138,7 +139,10 @@ func main() {
FedClient: federation,
KeyRing: keyRing,
AppserviceAPI: asAPI, FederationAPI: fsAPI,
AppserviceAPI: asAPI,
// always use the concrete impl here even in -http mode because adding public routes
// must be done on the concrete impl not an HTTP client else fedapi will call itself
FederationAPI: fsImplAPI,
RoomserverAPI: rsAPI,
UserAPI: userAPI,
KeyAPI: keyAPI,

4
cmd/dendrite-polylith-multi/main.go
View file

@ -71,8 +71,8 @@ func main() {
logrus.Infof("Starting %q component", component)
base := base.NewBaseDendrite(cfg, component) // TODO
defer base.Close() // nolint: errcheck
base := base.NewBaseDendrite(cfg, component, base.PolylithMode) // TODO
defer base.Close() // nolint: errcheck
go start(base, cfg)
base.WaitForShutdown()

76
federationapi/api/api.go
View file

@ -10,30 +10,6 @@ import (
"github.com/matrix-org/gomatrixserverlib"
)
// FederationClient is a subset of gomatrixserverlib.FederationClient functions which the fedsender
// implements as proxy calls, with built-in backoff/retries/etc. Errors returned from functions in
// this interface are of type FederationClientError
type FederationClient interface {
gomatrixserverlib.FederatedStateClient
GetUserDevices(ctx context.Context, s gomatrixserverlib.ServerName, userID string) (res gomatrixserverlib.RespUserDevices, err error)
ClaimKeys(ctx context.Context, s gomatrixserverlib.ServerName, oneTimeKeys map[string]map[string]string) (res gomatrixserverlib.RespClaimKeys, err error)
QueryKeys(ctx context.Context, s gomatrixserverlib.ServerName, keys map[string][]string) (res gomatrixserverlib.RespQueryKeys, err error)
MSC2836EventRelationships(ctx context.Context, dst gomatrixserverlib.ServerName, r gomatrixserverlib.MSC2836EventRelationshipsRequest, roomVersion gomatrixserverlib.RoomVersion) (res gomatrixserverlib.MSC2836EventRelationshipsResponse, err error)
MSC2946Spaces(ctx context.Context, dst gomatrixserverlib.ServerName, roomID string, suggestedOnly bool) (res gomatrixserverlib.MSC2946SpacesResponse, err error)
LookupServerKeys(ctx context.Context, s gomatrixserverlib.ServerName, keyRequests map[gomatrixserverlib.PublicKeyLookupRequest]gomatrixserverlib.Timestamp) ([]gomatrixserverlib.ServerKeys, error)
}
// FederationClientError is returned from FederationClient methods in the event of a problem.
type FederationClientError struct {
Err string
RetryAfter time.Duration
Blacklisted bool
}
func (e *FederationClientError) Error() string {
return fmt.Sprintf("%s - (retry_after=%s, blacklisted=%v)", e.Err, e.RetryAfter.String(), e.Blacklisted)
}
// FederationInternalAPI is used to query information from the federation sender.
type FederationInternalAPI interface {
FederationClient
@ -43,22 +19,7 @@ type FederationInternalAPI interface {
QueryServerKeys(ctx context.Context, request *QueryServerKeysRequest, response *QueryServerKeysResponse) error
// Query the server names of the joined hosts in a room.
// Unlike QueryJoinedHostsInRoom, this function returns a de-duplicated slice
// containing only the server names (without information for membership events).
// The response will include this server if they are joined to the room.
QueryJoinedHostServerNamesInRoom(
ctx context.Context,
request *QueryJoinedHostServerNamesInRoomRequest,
response *QueryJoinedHostServerNamesInRoomResponse,
) error
// Notifies the federation sender that these servers may be online and to retry sending messages.
PerformServersAlive(
ctx context.Context,
request *PerformServersAliveRequest,
response *PerformServersAliveResponse,
) error
// Broadcasts an EDU to all servers in rooms we are joined to.
// Broadcasts an EDU to all servers in rooms we are joined to. Used in the yggdrasil demos.
PerformBroadcastEDU(
ctx context.Context,
request *PerformBroadcastEDURequest,
@ -67,6 +28,10 @@ type FederationInternalAPI interface {
}
type ClientFederationAPI interface {
// Query the server names of the joined hosts in a room.
// Unlike QueryJoinedHostsInRoom, this function returns a de-duplicated slice
// containing only the server names (without information for membership events).
// The response will include this server if they are joined to the room.
QueryJoinedHostServerNamesInRoom(ctx context.Context, request *QueryJoinedHostServerNamesInRoomRequest, response *QueryJoinedHostServerNamesInRoomResponse) error
}
@ -95,6 +60,30 @@ type RoomserverFederationAPI interface {
LookupMissingEvents(ctx context.Context, s gomatrixserverlib.ServerName, roomID string, missing gomatrixserverlib.MissingEvents, roomVersion gomatrixserverlib.RoomVersion) (res gomatrixserverlib.RespMissingEvents, err error)
}
// FederationClient is a subset of gomatrixserverlib.FederationClient functions which the fedsender
// implements as proxy calls, with built-in backoff/retries/etc. Errors returned from functions in
// this interface are of type FederationClientError
type FederationClient interface {
gomatrixserverlib.FederatedStateClient
GetUserDevices(ctx context.Context, s gomatrixserverlib.ServerName, userID string) (res gomatrixserverlib.RespUserDevices, err error)
ClaimKeys(ctx context.Context, s gomatrixserverlib.ServerName, oneTimeKeys map[string]map[string]string) (res gomatrixserverlib.RespClaimKeys, err error)
QueryKeys(ctx context.Context, s gomatrixserverlib.ServerName, keys map[string][]string) (res gomatrixserverlib.RespQueryKeys, err error)
MSC2836EventRelationships(ctx context.Context, dst gomatrixserverlib.ServerName, r gomatrixserverlib.MSC2836EventRelationshipsRequest, roomVersion gomatrixserverlib.RoomVersion) (res gomatrixserverlib.MSC2836EventRelationshipsResponse, err error)
MSC2946Spaces(ctx context.Context, dst gomatrixserverlib.ServerName, roomID string, suggestedOnly bool) (res gomatrixserverlib.MSC2946SpacesResponse, err error)
LookupServerKeys(ctx context.Context, s gomatrixserverlib.ServerName, keyRequests map[gomatrixserverlib.PublicKeyLookupRequest]gomatrixserverlib.Timestamp) ([]gomatrixserverlib.ServerKeys, error)
}
// FederationClientError is returned from FederationClient methods in the event of a problem.
type FederationClientError struct {
Err string
RetryAfter time.Duration
Blacklisted bool
}
func (e *FederationClientError) Error() string {
return fmt.Sprintf("%s - (retry_after=%s, blacklisted=%v)", e.Err, e.RetryAfter.String(), e.Blacklisted)
}
type QueryServerKeysRequest struct {
ServerName gomatrixserverlib.ServerName
KeyIDToCriteria map[gomatrixserverlib.KeyID]gomatrixserverlib.PublicKeyNotaryQueryCriteria
@ -174,13 +163,6 @@ type PerformInviteResponse struct {
Event *gomatrixserverlib.HeaderedEvent `json:"event"`
}
type PerformServersAliveRequest struct {
Servers []gomatrixserverlib.ServerName
}
type PerformServersAliveResponse struct {
}
// QueryJoinedHostServerNamesInRoomRequest is a request to QueryJoinedHostServerNames
type QueryJoinedHostServerNamesInRoomRequest struct {
RoomID string `json:"room_id"`

21
federationapi/federationapi.go
View file

@ -50,13 +50,13 @@ func AddPublicRoutes(
federation *gomatrixserverlib.FederationClient,
keyRing gomatrixserverlib.JSONVerifier,
rsAPI roomserverAPI.FederationRoomserverAPI,
federationAPI federationAPI.FederationInternalAPI,
keyAPI keyserverAPI.KeyInternalAPI,
fedAPI federationAPI.FederationInternalAPI,
keyAPI keyserverAPI.FederationKeyAPI,
servers federationAPI.ServersInRoomProvider,
) {
cfg := &base.Cfg.FederationAPI
mscCfg := &base.Cfg.MSCs
js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
producer := &producers.SyncAPIProducer{
JetStream: js,
TopicReceiptEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
@ -67,12 +67,23 @@ func AddPublicRoutes(
UserAPI: userAPI,
}
// the federationapi component is a bit unique in that it attaches public routes AND serves
// internal APIs (because it used to be 2 components: the 2nd being fedsender). As a result,
// the constructor shape is a bit wonky in that it is not valid to AddPublicRoutes without a
// concrete impl of FederationInternalAPI as the public routes and the internal API _should_
// be the same thing now.
f, ok := fedAPI.(*internal.FederationInternalAPI)
if !ok {
panic("federationapi.AddPublicRoutes called with a FederationInternalAPI impl which was not " +
"FederationInternalAPI. This is a programming error.")
}
routing.Setup(
base.PublicFederationAPIMux,
base.PublicKeyAPIMux,
base.PublicWellKnownAPIMux,
cfg,
rsAPI, federationAPI, keyRing,
rsAPI, f, keyRing,
federation, userAPI, keyAPI, mscCfg,
servers, producer,
)
@ -104,7 +115,7 @@ func NewInternalAPI(
FailuresUntilBlacklist: cfg.FederationMaxRetries,
}
js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
queues := queue.NewOutgoingQueues(
federationDB, base.ProcessContext,

4
federationapi/federationapi_test.go
View file

@ -7,6 +7,7 @@ import (
"testing"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/internal"
"github.com/matrix-org/dendrite/internal/test"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
@ -27,10 +28,9 @@ func TestRoomsV3URLEscapeDoNot404(t *testing.T) {
cfg.FederationAPI.Database.ConnectionString = config.DataSource("file::memory:")
base := base.NewBaseDendrite(cfg, "Monolith")
keyRing := &test.NopJSONVerifier{}
fsAPI := base.FederationAPIHTTPClient()
// 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(base, nil, nil, keyRing, nil, fsAPI, nil, nil)
federationapi.AddPublicRoutes(base, nil, nil, keyRing, nil, &internal.FederationInternalAPI{}, nil, nil)
baseURL, cancel := test.ListenAndServe(t, base.PublicFederationAPIMux, true)
defer cancel()
serverName := gomatrixserverlib.ServerName(strings.TrimPrefix(baseURL, "https://"))

30
federationapi/internal/perform.go
View file

@ -563,20 +563,6 @@ func (r *FederationInternalAPI) PerformInvite(
return nil
}
// PerformServersAlive implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformServersAlive(
ctx context.Context,
request *api.PerformServersAliveRequest,
response *api.PerformServersAliveResponse,
) (err error) {
for _, srv := range request.Servers {
_ = r.db.RemoveServerFromBlacklist(srv)
r.queues.RetryServer(srv)
}
return nil
}
// PerformServersAlive implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformBroadcastEDU(
ctx context.Context,
@ -600,18 +586,18 @@ func (r *FederationInternalAPI) PerformBroadcastEDU(
if err = r.queues.SendEDU(edu, r.cfg.Matrix.ServerName, destinations); err != nil {
return fmt.Errorf("r.queues.SendEDU: %w", err)
}
wakeReq := &api.PerformServersAliveRequest{
Servers: destinations,
}
wakeRes := &api.PerformServersAliveResponse{}
if err := r.PerformServersAlive(ctx, wakeReq, wakeRes); err != nil {
return fmt.Errorf("r.PerformServersAlive: %w", err)
}
r.MarkServersAlive(destinations)
return nil
}
func (r *FederationInternalAPI) MarkServersAlive(destinations []gomatrixserverlib.ServerName) {
for _, srv := range destinations {
_ = r.db.RemoveServerFromBlacklist(srv)
r.queues.RetryServer(srv)
}
}
func sanityCheckAuthChain(authChain []*gomatrixserverlib.Event) error {
// sanity check we have a create event and it has a known room version
for _, ev := range authChain {

13
federationapi/inthttp/client.go
View file

@ -23,7 +23,6 @@ const (
FederationAPIPerformLeaveRequestPath = "/federationapi/performLeaveRequest"
FederationAPIPerformInviteRequestPath = "/federationapi/performInviteRequest"
FederationAPIPerformOutboundPeekRequestPath = "/federationapi/performOutboundPeekRequest"
FederationAPIPerformServersAlivePath = "/federationapi/performServersAlive"
FederationAPIPerformBroadcastEDUPath = "/federationapi/performBroadcastEDU"
FederationAPIGetUserDevicesPath = "/federationapi/client/getUserDevices"
@ -97,18 +96,6 @@ func (h *httpFederationInternalAPI) PerformOutboundPeek(
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
func (h *httpFederationInternalAPI) PerformServersAlive(
ctx context.Context,
request *api.PerformServersAliveRequest,
response *api.PerformServersAliveResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformServersAlive")
defer span.Finish()
apiURL := h.federationAPIURL + FederationAPIPerformServersAlivePath
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
// QueryJoinedHostServerNamesInRoom implements FederationInternalAPI
func (h *httpFederationInternalAPI) QueryJoinedHostServerNamesInRoom(
ctx context.Context,

14
federationapi/inthttp/server.go
View file

@ -81,20 +81,6 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
internalAPIMux.Handle(
FederationAPIPerformServersAlivePath,
httputil.MakeInternalAPI("PerformServersAliveRequest", func(req *http.Request) util.JSONResponse {
var request api.PerformServersAliveRequest
var response api.PerformServersAliveResponse
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
return util.MessageResponse(http.StatusBadRequest, err.Error())
}
if err := intAPI.PerformServersAlive(req.Context(), &request, &response); err != nil {
return util.ErrorResponse(err)
}
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
internalAPIMux.Handle(
FederationAPIPerformBroadcastEDUPath,
httputil.MakeInternalAPI("PerformBroadcastEDU", func(req *http.Request) util.JSONResponse {

2
federationapi/routing/devices.go
View file

@ -26,7 +26,7 @@ import (
// GetUserDevices for the given user id
func GetUserDevices(
req *http.Request,
keyAPI keyapi.KeyInternalAPI,
keyAPI keyapi.FederationKeyAPI,
userID string,
) util.JSONResponse {
var res keyapi.QueryDeviceMessagesResponse

4
federationapi/routing/keys.go
View file

@ -37,7 +37,7 @@ type queryKeysRequest struct {
// QueryDeviceKeys returns device keys for users on this server.
// https://matrix.org/docs/spec/server_server/latest#post-matrix-federation-v1-user-keys-query
func QueryDeviceKeys(
httpReq *http.Request, request *gomatrixserverlib.FederationRequest, keyAPI api.KeyInternalAPI, thisServer gomatrixserverlib.ServerName,
httpReq *http.Request, request *gomatrixserverlib.FederationRequest, keyAPI api.FederationKeyAPI, thisServer gomatrixserverlib.ServerName,
) util.JSONResponse {
var qkr queryKeysRequest
err := json.Unmarshal(request.Content(), &qkr)
@ -89,7 +89,7 @@ type claimOTKsRequest struct {
// ClaimOneTimeKeys claims OTKs for users on this server.
// https://matrix.org/docs/spec/server_server/latest#post-matrix-federation-v1-user-keys-claim
func ClaimOneTimeKeys(
httpReq *http.Request, request *gomatrixserverlib.FederationRequest, keyAPI api.KeyInternalAPI, thisServer gomatrixserverlib.ServerName,
httpReq *http.Request, request *gomatrixserverlib.FederationRequest, keyAPI api.FederationKeyAPI, thisServer gomatrixserverlib.ServerName,
) util.JSONResponse {
var cor claimOTKsRequest
err := json.Unmarshal(request.Content(), &cor)

118
federationapi/routing/routing.go
View file

@ -18,10 +18,14 @@ import (
"context"
"fmt"
"net/http"
"sync"
"time"
"github.com/getsentry/sentry-go"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
fedInternal "github.com/matrix-org/dendrite/federationapi/internal"
"github.com/matrix-org/dendrite/federationapi/producers"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
@ -48,11 +52,11 @@ func Setup(
fedMux, keyMux, wkMux *mux.Router,
cfg *config.FederationAPI,
rsAPI roomserverAPI.FederationRoomserverAPI,
fsAPI federationAPI.FederationInternalAPI,
fsAPI *fedInternal.FederationInternalAPI,
keys gomatrixserverlib.JSONVerifier,
federation *gomatrixserverlib.FederationClient,
userAPI userapi.FederationUserAPI,
keyAPI keyserverAPI.KeyInternalAPI,
keyAPI keyserverAPI.FederationKeyAPI,
mscCfg *config.MSCs,
servers federationAPI.ServersInRoomProvider,
producer *producers.SyncAPIProducer,
@ -65,7 +69,7 @@ func Setup(
v1fedmux := fedMux.PathPrefix("/v1").Subrouter()
v2fedmux := fedMux.PathPrefix("/v2").Subrouter()
wakeup := &httputil.FederationWakeups{
wakeup := &FederationWakeups{
FsAPI: fsAPI,
}
@ -119,7 +123,7 @@ func Setup(
v2keysmux.Handle("/query/{serverName}/{keyID}", notaryKeys).Methods(http.MethodGet)
mu := internal.NewMutexByRoom()
v1fedmux.Handle("/send/{txnID}", httputil.MakeFedAPI(
v1fedmux.Handle("/send/{txnID}", MakeFedAPI(
"federation_send", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return Send(
@ -129,7 +133,7 @@ func Setup(
},
)).Methods(http.MethodPut, http.MethodOptions)
v1fedmux.Handle("/invite/{roomID}/{eventID}", httputil.MakeFedAPI(
v1fedmux.Handle("/invite/{roomID}/{eventID}", MakeFedAPI(
"federation_invite", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -145,7 +149,7 @@ func Setup(
},
)).Methods(http.MethodPut, http.MethodOptions)
v2fedmux.Handle("/invite/{roomID}/{eventID}", httputil.MakeFedAPI(
v2fedmux.Handle("/invite/{roomID}/{eventID}", MakeFedAPI(
"federation_invite", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -167,7 +171,7 @@ func Setup(
},
)).Methods(http.MethodPost, http.MethodOptions)
v1fedmux.Handle("/exchange_third_party_invite/{roomID}", httputil.MakeFedAPI(
v1fedmux.Handle("/exchange_third_party_invite/{roomID}", MakeFedAPI(
"exchange_third_party_invite", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return ExchangeThirdPartyInvite(
@ -176,7 +180,7 @@ func Setup(
},
)).Methods(http.MethodPut, http.MethodOptions)
v1fedmux.Handle("/event/{eventID}", httputil.MakeFedAPI(
v1fedmux.Handle("/event/{eventID}", MakeFedAPI(
"federation_get_event", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return GetEvent(
@ -185,7 +189,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/state/{roomID}", httputil.MakeFedAPI(
v1fedmux.Handle("/state/{roomID}", MakeFedAPI(
"federation_get_state", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -200,7 +204,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/state_ids/{roomID}", httputil.MakeFedAPI(
v1fedmux.Handle("/state_ids/{roomID}", MakeFedAPI(
"federation_get_state_ids", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -215,7 +219,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/event_auth/{roomID}/{eventID}", httputil.MakeFedAPI(
v1fedmux.Handle("/event_auth/{roomID}/{eventID}", MakeFedAPI(
"federation_get_event_auth", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -230,7 +234,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/query/directory", httputil.MakeFedAPI(
v1fedmux.Handle("/query/directory", MakeFedAPI(
"federation_query_room_alias", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return RoomAliasToID(
@ -239,7 +243,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/query/profile", httputil.MakeFedAPI(
v1fedmux.Handle("/query/profile", MakeFedAPI(
"federation_query_profile", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return GetProfile(
@ -248,7 +252,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/user/devices/{userID}", httputil.MakeFedAPI(
v1fedmux.Handle("/user/devices/{userID}", MakeFedAPI(
"federation_user_devices", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return GetUserDevices(
@ -258,7 +262,7 @@ func Setup(
)).Methods(http.MethodGet)
if mscCfg.Enabled("msc2444") {
v1fedmux.Handle("/peek/{roomID}/{peekID}", httputil.MakeFedAPI(
v1fedmux.Handle("/peek/{roomID}/{peekID}", MakeFedAPI(
"federation_peek", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -289,7 +293,7 @@ func Setup(
)).Methods(http.MethodPut, http.MethodDelete)
}
v1fedmux.Handle("/make_join/{roomID}/{userID}", httputil.MakeFedAPI(
v1fedmux.Handle("/make_join/{roomID}/{userID}", MakeFedAPI(
"federation_make_join", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -320,7 +324,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/send_join/{roomID}/{eventID}", httputil.MakeFedAPI(
v1fedmux.Handle("/send_join/{roomID}/{eventID}", MakeFedAPI(
"federation_send_join", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -352,7 +356,7 @@ func Setup(
},
)).Methods(http.MethodPut)
v2fedmux.Handle("/send_join/{roomID}/{eventID}", httputil.MakeFedAPI(
v2fedmux.Handle("/send_join/{roomID}/{eventID}", MakeFedAPI(
"federation_send_join", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -369,7 +373,7 @@ func Setup(
},
)).Methods(http.MethodPut)
v1fedmux.Handle("/make_leave/{roomID}/{eventID}", httputil.MakeFedAPI(
v1fedmux.Handle("/make_leave/{roomID}/{eventID}", MakeFedAPI(
"federation_make_leave", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -386,7 +390,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/send_leave/{roomID}/{eventID}", httputil.MakeFedAPI(
v1fedmux.Handle("/send_leave/{roomID}/{eventID}", MakeFedAPI(
"federation_send_leave", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -418,7 +422,7 @@ func Setup(
},
)).Methods(http.MethodPut)
v2fedmux.Handle("/send_leave/{roomID}/{eventID}", httputil.MakeFedAPI(
v2fedmux.Handle("/send_leave/{roomID}/{eventID}", MakeFedAPI(
"federation_send_leave", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -442,7 +446,7 @@ func Setup(
},
)).Methods(http.MethodGet)
v1fedmux.Handle("/get_missing_events/{roomID}", httputil.MakeFedAPI(
v1fedmux.Handle("/get_missing_events/{roomID}", MakeFedAPI(
"federation_get_missing_events", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -455,7 +459,7 @@ func Setup(
},
)).Methods(http.MethodPost)
v1fedmux.Handle("/backfill/{roomID}", httputil.MakeFedAPI(
v1fedmux.Handle("/backfill/{roomID}", MakeFedAPI(
"federation_backfill", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@ -474,14 +478,14 @@ func Setup(
}),
).Methods(http.MethodGet, http.MethodPost)
v1fedmux.Handle("/user/keys/claim", httputil.MakeFedAPI(
v1fedmux.Handle("/user/keys/claim", MakeFedAPI(
"federation_keys_claim", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return ClaimOneTimeKeys(httpReq, request, keyAPI, cfg.Matrix.ServerName)
},
)).Methods(http.MethodPost)
v1fedmux.Handle("/user/keys/query", httputil.MakeFedAPI(
v1fedmux.Handle("/user/keys/query", MakeFedAPI(
"federation_keys_query", cfg.Matrix.ServerName, keys, wakeup,
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return QueryDeviceKeys(httpReq, request, keyAPI, cfg.Matrix.ServerName)
@ -518,3 +522,67 @@ func ErrorIfLocalServerNotInRoom(
}
return nil
}
// MakeFedAPI makes an http.Handler that checks matrix federation authentication.
func MakeFedAPI(
metricsName string,
serverName gomatrixserverlib.ServerName,
keyRing gomatrixserverlib.JSONVerifier,
wakeup *FederationWakeups,
f func(*http.Request, *gomatrixserverlib.FederationRequest, map[string]string) util.JSONResponse,
) http.Handler {
h := func(req *http.Request) util.JSONResponse {
fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest(
req, time.Now(), serverName, keyRing,
)
if fedReq == nil {
return errResp
}
// add the user to Sentry, if enabled
hub := sentry.GetHubFromContext(req.Context())
if hub != nil {
hub.Scope().SetTag("origin", string(fedReq.Origin()))
hub.Scope().SetTag("uri", fedReq.RequestURI())
}
defer func() {
if r := recover(); r != nil {
if hub != nil {
hub.CaptureException(fmt.Errorf("%s panicked", req.URL.Path))
}
// re-panic to return the 500
panic(r)
}
}()
go wakeup.Wakeup(req.Context(), fedReq.Origin())
vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
if err != nil {
return util.MatrixErrorResponse(400, "M_UNRECOGNISED", "badly encoded query params")
}
jsonRes := f(req, fedReq, vars)
// do not log 4xx as errors as they are client fails, not server fails
if hub != nil && jsonRes.Code >= 500 {
hub.Scope().SetExtra("response", jsonRes)
hub.CaptureException(fmt.Errorf("%s returned HTTP %d", req.URL.Path, jsonRes.Code))
}
return jsonRes
}
return httputil.MakeExternalAPI(metricsName, h)
}
type FederationWakeups struct {
FsAPI *fedInternal.FederationInternalAPI
origins sync.Map
}
func (f *FederationWakeups) Wakeup(ctx context.Context, origin gomatrixserverlib.ServerName) {
key, keyok := f.origins.Load(origin)
if keyok {
lastTime, ok := key.(time.Time)
if ok && time.Since(lastTime) < time.Minute {
return
}
}
f.FsAPI.MarkServersAlive([]gomatrixserverlib.ServerName{origin})
f.origins.Store(origin, time.Now())
}

4
federationapi/routing/send.go
View file

@ -83,7 +83,7 @@ func Send(
txnID gomatrixserverlib.TransactionID,
cfg *config.FederationAPI,
rsAPI api.FederationRoomserverAPI,
keyAPI keyapi.KeyInternalAPI,
keyAPI keyapi.FederationKeyAPI,
keys gomatrixserverlib.JSONVerifier,
federation *gomatrixserverlib.FederationClient,
mu *internal.MutexByRoom,
@ -183,7 +183,7 @@ func Send(
type txnReq struct {
gomatrixserverlib.Transaction
rsAPI api.FederationRoomserverAPI
keyAPI keyapi.KeyInternalAPI
keyAPI keyapi.FederationKeyAPI
ourServerName gomatrixserverlib.ServerName
keys gomatrixserverlib.JSONVerifier
federation txnFederationClient

8
go.mod
View file

@ -30,7 +30,7 @@ require (
github.com/matrix-org/dugong v0.0.0-20210921133753-66e6b1c67e2e
github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91
github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16
github.com/matrix-org/gomatrixserverlib v0.0.0-20220505130352-f72a63510060
github.com/matrix-org/gomatrixserverlib v0.0.0-20220509120958-8d818048c34c
github.com/matrix-org/pinecone v0.0.0-20220408153826-2999ea29ed48
github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
github.com/mattn/go-sqlite3 v1.14.10
@ -48,17 +48,17 @@ require (
github.com/prometheus/client_golang v1.12.1
github.com/sirupsen/logrus v1.8.1
github.com/stretchr/testify v1.7.0
github.com/tidwall/gjson v1.14.0
github.com/tidwall/gjson v1.14.1
github.com/tidwall/sjson v1.2.4
github.com/uber/jaeger-client-go v2.30.0+incompatible
github.com/uber/jaeger-lib v2.4.1+incompatible
github.com/yggdrasil-network/yggdrasil-go v0.4.3
go.uber.org/atomic v1.9.0
golang.org/x/crypto v0.0.0-20220331220935-ae2d96664a29
golang.org/x/crypto v0.0.0-20220507011949-2cf3adece122
golang.org/x/image v0.0.0-20220321031419-a8550c1d254a
golang.org/x/mobile v0.0.0-20220407111146-e579adbbc4a2
golang.org/x/net v0.0.0-20220407224826-aac1ed45d8e3
golang.org/x/sys v0.0.0-20220406163625-3f8b81556e12 // indirect
golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6 // indirect
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
gopkg.in/h2non/bimg.v1 v1.1.9
gopkg.in/yaml.v2 v2.4.0

16
go.sum
View file

@ -795,8 +795,8 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91/go.mod h1
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 h1:ZtO5uywdd5dLDCud4r0r55eP4j9FuUNpl60Gmntcop4=
github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220505130352-f72a63510060 h1:tYi4mCOWgVLt8mpkG1LFRKcMfSTwp5NQ5wBKdtaxO9s=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220505130352-f72a63510060/go.mod h1:V5eO8rn/C3rcxig37A/BCeKerLFS+9Avg/77FIeTZ48=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220509120958-8d818048c34c h1:KqzqFWxvs90pcDaW9QEveW+Q5JcEYuNnKyaqXc+ohno=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220509120958-8d818048c34c/go.mod h1:V5eO8rn/C3rcxig37A/BCeKerLFS+9Avg/77FIeTZ48=
github.com/matrix-org/pinecone v0.0.0-20220408153826-2999ea29ed48 h1:W0sjjC6yjskHX4mb0nk3p0fXAlbU5bAFUFeEtlrPASE=
github.com/matrix-org/pinecone v0.0.0-20220408153826-2999ea29ed48/go.mod h1:ulJzsVOTssIVp1j/m5eI//4VpAGDkMt5NrRuAVX7wpc=
github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U=
@ -1130,8 +1130,8 @@ github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG
github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA=
github.com/tchap/go-patricia v2.2.6+incompatible/go.mod h1:bmLyhP68RS6kStMGxByiQ23RP/odRBOTVjwp2cDyi6I=
github.com/tidwall/gjson v1.12.1/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
github.com/tidwall/gjson v1.14.0 h1:6aeJ0bzojgWLa82gDQHcx3S0Lr/O51I9bJ5nv6JFx5w=
github.com/tidwall/gjson v1.14.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
github.com/tidwall/gjson v1.14.1 h1:iymTbGkQBhveq21bEvAQ81I0LEBork8BFe1CUZXdyuo=
github.com/tidwall/gjson v1.14.1/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
@ -1281,8 +1281,8 @@ golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm
golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/crypto v0.0.0-20220331220935-ae2d96664a29 h1:tkVvjkPTB7pnW3jnid7kNyAMPVWllTNOf/qKDze4p9o=
golang.org/x/crypto v0.0.0-20220331220935-ae2d96664a29/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/crypto v0.0.0-20220507011949-2cf3adece122 h1:NvGWuYG8dkDHFSKksI1P9faiVJ9rayE6l0+ouWVIDs8=
golang.org/x/crypto v0.0.0-20220507011949-2cf3adece122/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@ -1543,8 +1543,8 @@ golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220405052023-b1e9470b6e64/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220406163625-3f8b81556e12 h1:QyVthZKMsyaQwBTJE04jdNN0Pp5Fn9Qga0mrgxyERQM=
golang.org/x/sys v0.0.0-20220406163625-3f8b81556e12/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6 h1:nonptSpoQ4vQjyraW20DXPAglgQfVnM9ZC6MmNLMR60=
golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

35
internal/caching/cache_lazy_load_members.go
View file

@ -15,33 +15,14 @@ const (
LazyLoadCacheMaxAge = time.Minute * 30
)
type LazyLoadCache struct {
// InMemoryLRUCachePartition containing other InMemoryLRUCachePartitions
// with the actual cached members
userCaches *InMemoryLRUCachePartition
type LazyLoadCache interface {
StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string)
IsLazyLoadedUserCached(device *userapi.Device, roomID, userID string) (string, bool)
}
// NewLazyLoadCache creates a new LazyLoadCache.
func NewLazyLoadCache() (*LazyLoadCache, error) {
cache, err := NewInMemoryLRUCachePartition(
LazyLoadCacheName,
LazyLoadCacheMutable,
LazyLoadCacheMaxEntries,
LazyLoadCacheMaxAge,
true,
)
if err != nil {
return nil, err
}
go cacheCleaner(cache)
return &LazyLoadCache{
userCaches: cache,
}, nil
}
func (c *LazyLoadCache) lazyLoadCacheForUser(device *userapi.Device) (*InMemoryLRUCachePartition, error) {
func (c Caches) lazyLoadCacheForUser(device *userapi.Device) (*InMemoryLRUCachePartition, error) {
cacheName := fmt.Sprintf("%s/%s", device.UserID, device.ID)
userCache, ok := c.userCaches.Get(cacheName)
userCache, ok := c.LazyLoading.Get(cacheName)
if ok && userCache != nil {
if cache, ok := userCache.(*InMemoryLRUCachePartition); ok {
return cache, nil
@ -57,12 +38,12 @@ func (c *LazyLoadCache) lazyLoadCacheForUser(device *userapi.Device) (*InMemoryL
if err != nil {
return nil, err
}
c.userCaches.Set(cacheName, cache)
c.LazyLoading.Set(cacheName, cache)
go cacheCleaner(cache)
return cache, nil
}
func (c *LazyLoadCache) StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string) {
func (c Caches) StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string) {
cache, err := c.lazyLoadCacheForUser(device)
if err != nil {
return
@ -71,7 +52,7 @@ func (c *LazyLoadCache) StoreLazyLoadedUser(device *userapi.Device, roomID, user
cache.Set(cacheKey, eventID)
}
func (c *LazyLoadCache) IsLazyLoadedUserCached(device *userapi.Device, roomID, userID string) (string, bool) {
func (c Caches) IsLazyLoadedUserCached(device *userapi.Device, roomID, userID string) (string, bool) {
cache, err := c.lazyLoadCacheForUser(device)
if err != nil {
return "", false

5
internal/caching/caches.go
View file

@ -1,6 +1,8 @@
package caching
import "time"
import (
"time"
)
// Caches contains a set of references to caches. They may be
// different implementations as long as they satisfy the Cache
@ -13,6 +15,7 @@ type Caches struct {
RoomInfos Cache // RoomInfoCache
FederationEvents Cache // FederationEventsCache
SpaceSummaryRooms Cache // SpaceSummaryRoomsCache
LazyLoading Cache // LazyLoadCache
}
// Cache is the interface that an implementation must satisfy.

15
internal/caching/impl_inmemorylru.go
View file

@ -70,9 +70,21 @@ func NewInMemoryLRUCache(enablePrometheus bool) (*Caches, error) {
if err != nil {
return nil, err
}
lazyLoadCache, err := NewInMemoryLRUCachePartition(
LazyLoadCacheName,
LazyLoadCacheMutable,
LazyLoadCacheMaxEntries,
LazyLoadCacheMaxAge,
enablePrometheus,
)
if err != nil {
return nil, err
}
go cacheCleaner(
roomVersions, serverKeys, roomServerRoomIDs,
roomInfos, federationEvents, spaceRooms,
roomInfos, federationEvents, spaceRooms, lazyLoadCache,
)
return &Caches{
RoomVersions: roomVersions,
@ -81,6 +93,7 @@ func NewInMemoryLRUCache(enablePrometheus bool) (*Caches, error) {
RoomInfos: roomInfos,
FederationEvents: federationEvents,
SpaceSummaryRooms: spaceRooms,
LazyLoading: lazyLoadCache,
}, nil
}

77
internal/httputil/httpapi.go
View file

@ -24,17 +24,13 @@ import (
"net/http/httputil"
"os"
"strings"
"sync"
"time"
"github.com/getsentry/sentry-go"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/clientapi/auth"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
federationapiAPI "github.com/matrix-org/dendrite/federationapi/api"
"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"
opentracing "github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go/ext"
@ -298,79 +294,6 @@ func MakeInternalAPI(metricsName string, f func(*http.Request) util.JSONResponse
)
}
// MakeFedAPI makes an http.Handler that checks matrix federation authentication.
func MakeFedAPI(
metricsName string,
serverName gomatrixserverlib.ServerName,
keyRing gomatrixserverlib.JSONVerifier,
wakeup *FederationWakeups,
f func(*http.Request, *gomatrixserverlib.FederationRequest, map[string]string) util.JSONResponse,
) http.Handler {
h := func(req *http.Request) util.JSONResponse {
fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest(
req, time.Now(), serverName, keyRing,
)
if fedReq == nil {
return errResp
}
// add the user to Sentry, if enabled
hub := sentry.GetHubFromContext(req.Context())
if hub != nil {
hub.Scope().SetTag("origin", string(fedReq.Origin()))
hub.Scope().SetTag("uri", fedReq.RequestURI())
}
defer func() {
if r := recover(); r != nil {
if hub != nil {
hub.CaptureException(fmt.Errorf("%s panicked", req.URL.Path))
}
// re-panic to return the 500
panic(r)
}
}()
go wakeup.Wakeup(req.Context(), fedReq.Origin())
vars, err := URLDecodeMapValues(mux.Vars(req))
if err != nil {
return util.MatrixErrorResponse(400, "M_UNRECOGNISED", "badly encoded query params")
}
jsonRes := f(req, fedReq, vars)
// do not log 4xx as errors as they are client fails, not server fails
if hub != nil && jsonRes.Code >= 500 {
hub.Scope().SetExtra("response", jsonRes)
hub.CaptureException(fmt.Errorf("%s returned HTTP %d", req.URL.Path, jsonRes.Code))
}
return jsonRes
}
return MakeExternalAPI(metricsName, h)
}
type FederationWakeups struct {
FsAPI federationapiAPI.FederationInternalAPI
origins sync.Map
}
func (f *FederationWakeups) Wakeup(ctx context.Context, origin gomatrixserverlib.ServerName) {
key, keyok := f.origins.Load(origin)
if keyok {
lastTime, ok := key.(time.Time)
if ok && time.Since(lastTime) < time.Minute {
return
}
}
aliveReq := federationapiAPI.PerformServersAliveRequest{
Servers: []gomatrixserverlib.ServerName{origin},
}
aliveRes := federationapiAPI.PerformServersAliveResponse{}
if err := f.FsAPI.PerformServersAlive(ctx, &aliveReq, &aliveRes); err != nil {
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
"origin": origin,
}).Warn("incoming federation request failed to notify server alive")
} else {
f.origins.Store(origin, time.Now())
}
}
// WrapHandlerInBasicAuth adds basic auth to a handler. Only used for /metrics
func WrapHandlerInBasicAuth(h http.Handler, b BasicAuth) http.HandlerFunc {
if b.Username == "" || b.Password == "" {

4
internal/version.go
View file

@ -17,8 +17,8 @@ var build string
const (
VersionMajor = 0
VersionMinor = 8
VersionPatch = 2
VersionTag = "" // example: "rc1"
VersionPatch = 3
VersionTag = "rc1" // example: "rc1"
)
func VersionString() string {

18
keyserver/api/api.go
View file

@ -29,15 +29,11 @@ import (
type KeyInternalAPI interface {
SyncKeyAPI
ClientKeyAPI
FederationKeyAPI
UserKeyAPI
// SetUserAPI assigns a user API to query when extracting device names.
SetUserAPI(i userapi.UserInternalAPI)
// InputDeviceListUpdate from a federated server EDU
InputDeviceListUpdate(ctx context.Context, req *InputDeviceListUpdateRequest, res *InputDeviceListUpdateResponse)
QueryDeviceMessages(ctx context.Context, req *QueryDeviceMessagesRequest, res *QueryDeviceMessagesResponse)
QuerySignatures(ctx context.Context, req *QuerySignaturesRequest, res *QuerySignaturesResponse)
SetUserAPI(i userapi.KeyserverUserAPI)
}
// API functions required by the clientapi
@ -62,6 +58,16 @@ type SyncKeyAPI interface {
QueryOneTimeKeys(ctx context.Context, req *QueryOneTimeKeysRequest, res *QueryOneTimeKeysResponse)
}
type FederationKeyAPI interface {
QueryKeys(ctx context.Context, req *QueryKeysRequest, res *QueryKeysResponse)
QuerySignatures(ctx context.Context, req *QuerySignaturesRequest, res *QuerySignaturesResponse)
QueryDeviceMessages(ctx context.Context, req *QueryDeviceMessagesRequest, res *QueryDeviceMessagesResponse)
// InputDeviceListUpdate from a federated server EDU
InputDeviceListUpdate(ctx context.Context, req *InputDeviceListUpdateRequest, res *InputDeviceListUpdateResponse)
PerformUploadDeviceKeys(ctx context.Context, req *PerformUploadDeviceKeysRequest, res *PerformUploadDeviceKeysResponse)
PerformClaimKeys(ctx context.Context, req *PerformClaimKeysRequest, res *PerformClaimKeysResponse)
}
// KeyError is returned if there was a problem performing/querying the server
type KeyError struct {
Err string `json:"error"`

4
keyserver/internal/internal.go
View file

@ -38,12 +38,12 @@ type KeyInternalAPI struct {
DB storage.Database
ThisServer gomatrixserverlib.ServerName
FedClient fedsenderapi.FederationClient
UserAPI userapi.UserInternalAPI
UserAPI userapi.KeyserverUserAPI
Producer *producers.KeyChange
Updater *DeviceListUpdater
}
func (a *KeyInternalAPI) SetUserAPI(i userapi.UserInternalAPI) {
func (a *KeyInternalAPI) SetUserAPI(i userapi.KeyserverUserAPI) {
a.UserAPI = i
}

2
keyserver/inthttp/client.go
View file

@ -60,7 +60,7 @@ type httpKeyInternalAPI struct {
httpClient *http.Client
}
func (h *httpKeyInternalAPI) SetUserAPI(i userapi.UserInternalAPI) {
func (h *httpKeyInternalAPI) SetUserAPI(i userapi.KeyserverUserAPI) {
// no-op: doesn't need it
}
func (h *httpKeyInternalAPI) InputDeviceListUpdate(

2
keyserver/keyserver.go
View file

@ -39,7 +39,7 @@ func AddInternalRoutes(router *mux.Router, intAPI api.KeyInternalAPI) {
func NewInternalAPI(
base *base.BaseDendrite, cfg *config.KeyServer, fedClient fedsenderapi.FederationClient,
) api.KeyInternalAPI {
js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
db, err := storage.NewDatabase(base, &cfg.Database)
if err != nil {

2
keyserver/storage/postgres/one_time_keys_table.go
View file

@ -39,6 +39,8 @@ CREATE TABLE IF NOT EXISTS keyserver_one_time_keys (
-- Clobber based on 4-uple of user/device/key/algorithm.
CONSTRAINT keyserver_one_time_keys_unique UNIQUE (user_id, device_id, key_id, algorithm)
);
CREATE INDEX IF NOT EXISTS keyserver_one_time_keys_idx ON keyserver_one_time_keys (user_id, device_id);
`
const upsertKeysSQL = "" +

2
keyserver/storage/sqlite3/one_time_keys_table.go
View file

@ -38,6 +38,8 @@ CREATE TABLE IF NOT EXISTS keyserver_one_time_keys (
-- Clobber based on 4-uple of user/device/key/algorithm.
UNIQUE (user_id, device_id, key_id, algorithm)
);
CREATE INDEX IF NOT EXISTS keyserver_one_time_keys_idx ON keyserver_one_time_keys (user_id, device_id);
`
const upsertKeysSQL = "" +

12
roomserver/api/alias.go
View file

@ -59,18 +59,6 @@ type GetAliasesForRoomIDResponse struct {
Aliases []string `json:"aliases"`
}
// GetCreatorIDForAliasRequest is a request to GetCreatorIDForAlias
type GetCreatorIDForAliasRequest struct {
// The alias we want to find the creator of
Alias string `json:"alias"`
}
// GetCreatorIDForAliasResponse is a response to GetCreatorIDForAlias
type GetCreatorIDForAliasResponse struct {
// The user ID of the alias creator
UserID string `json:"user_id"`
}
// RemoveRoomAliasRequest is a request to RemoveRoomAlias
type RemoveRoomAliasRequest struct {
// ID of the user removing the alias

93
roomserver/api/api.go
View file

@ -12,10 +12,6 @@ import (
// RoomserverInputAPI is used to write events to the room server.
type RoomserverInternalAPI interface {
InputRoomEventsAPI
QueryLatestEventsAndStateAPI
QueryEventsAPI
SyncRoomserverAPI
AppserviceRoomserverAPI
ClientRoomserverAPI
@ -25,101 +21,18 @@ type RoomserverInternalAPI interface {
// needed to avoid chicken and egg scenario when setting up the
// interdependencies between the roomserver and other input APIs
SetFederationAPI(fsAPI fsAPI.RoomserverFederationAPI, keyRing *gomatrixserverlib.KeyRing)
SetAppserviceAPI(asAPI asAPI.AppServiceQueryAPI)
SetUserAPI(userAPI userapi.UserInternalAPI)
PerformInvite(
ctx context.Context,
req *PerformInviteRequest,
res *PerformInviteResponse,
) error
PerformJoin(
ctx context.Context,
req *PerformJoinRequest,
res *PerformJoinResponse,
)
PerformLeave(
ctx context.Context,
req *PerformLeaveRequest,
res *PerformLeaveResponse,
) error
PerformPeek(
ctx context.Context,
req *PerformPeekRequest,
res *PerformPeekResponse,
)
PerformUnpeek(
ctx context.Context,
req *PerformUnpeekRequest,
res *PerformUnpeekResponse,
)
PerformPublish(
ctx context.Context,
req *PerformPublishRequest,
res *PerformPublishResponse,
)
SetAppserviceAPI(asAPI asAPI.AppServiceInternalAPI)
SetUserAPI(userAPI userapi.RoomserverUserAPI)
// QueryAuthChain returns the entire auth chain for the event IDs given.
// The response includes the events in the request.
// Omits without error for any missing auth events. There will be no duplicates.
// Used in MSC2836.
QueryAuthChain(
ctx context.Context,
req *QueryAuthChainRequest,
res *QueryAuthChainResponse,
) error
// QueryRoomsForUser retrieves a list of room IDs matching the given query.
QueryRoomsForUser(ctx context.Context, req *QueryRoomsForUserRequest, res *QueryRoomsForUserResponse) error
// PerformRoomUpgrade upgrades a room to a newer version
PerformRoomUpgrade(ctx context.Context, req *PerformRoomUpgradeRequest, resp *PerformRoomUpgradeResponse)
// Asks for the default room version as preferred by the server.
QueryRoomVersionCapabilities(
ctx context.Context,
req *QueryRoomVersionCapabilitiesRequest,
res *QueryRoomVersionCapabilitiesResponse,
) error
// Asks for the room version for a given room.
QueryRoomVersionForRoom(
ctx context.Context,
req *QueryRoomVersionForRoomRequest,
res *QueryRoomVersionForRoomResponse,
) error
// Set a room alias
SetRoomAlias(
ctx context.Context,
req *SetRoomAliasRequest,
res *SetRoomAliasResponse,
) error
// Get the room ID for an alias
GetRoomIDForAlias(
ctx context.Context,
req *GetRoomIDForAliasRequest,
res *GetRoomIDForAliasResponse,
) error
// Get the user ID of the creator of an alias
GetCreatorIDForAlias(
ctx context.Context,
req *GetCreatorIDForAliasRequest,
res *GetCreatorIDForAliasResponse,
) error
// Remove a room alias
RemoveRoomAlias(
ctx context.Context,
req *RemoveRoomAliasRequest,
res *RemoveRoomAliasResponse,
) error
}
type InputRoomEventsAPI interface {

14
roomserver/api/api_trace.go
View file

@ -23,11 +23,11 @@ func (t *RoomserverInternalAPITrace) SetFederationAPI(fsAPI fsAPI.RoomserverFede
t.Impl.SetFederationAPI(fsAPI, keyRing)
}
func (t *RoomserverInternalAPITrace) SetAppserviceAPI(asAPI asAPI.AppServiceQueryAPI) {
func (t *RoomserverInternalAPITrace) SetAppserviceAPI(asAPI asAPI.AppServiceInternalAPI) {
t.Impl.SetAppserviceAPI(asAPI)
}
func (t *RoomserverInternalAPITrace) SetUserAPI(userAPI userapi.UserInternalAPI) {
func (t *RoomserverInternalAPITrace) SetUserAPI(userAPI userapi.RoomserverUserAPI) {
t.Impl.SetUserAPI(userAPI)
}
@ -293,16 +293,6 @@ func (t *RoomserverInternalAPITrace) GetAliasesForRoomID(
return err
}
func (t *RoomserverInternalAPITrace) GetCreatorIDForAlias(
ctx context.Context,
req *GetCreatorIDForAliasRequest,
res *GetCreatorIDForAliasResponse,
) error {
err := t.Impl.GetCreatorIDForAlias(ctx, req, res)
util.GetLogger(ctx).WithError(err).Infof("GetCreatorIDForAlias req=%+v res=%+v", js(req), js(res))
return err
}
func (t *RoomserverInternalAPITrace) RemoveRoomAlias(
ctx context.Context,
req *RemoveRoomAliasRequest,

19
roomserver/internal/alias.go
View file

@ -41,9 +41,6 @@ type RoomserverInternalAPIDatabase interface {
// Look up all aliases referring to a given room ID.
// Returns an error if there was a problem talking to the database.
GetAliasesForRoomID(ctx context.Context, roomID string) ([]string, error)
// Get the user ID of the creator of an alias.
// Returns an error if there was a problem talking to the database.
GetCreatorIDForAlias(ctx context.Context, alias string) (string, error)
// Remove a given room alias.
// Returns an error if there was a problem talking to the database.
RemoveRoomAlias(ctx context.Context, alias string) error
@ -134,22 +131,6 @@ func (r *RoomserverInternalAPI) GetAliasesForRoomID(
return nil
}
// GetCreatorIDForAlias implements alias.RoomserverInternalAPI
func (r *RoomserverInternalAPI) GetCreatorIDForAlias(
ctx context.Context,
request *api.GetCreatorIDForAliasRequest,
response *api.GetCreatorIDForAliasResponse,
) error {
// Look up the aliases in the database for the given RoomID
creatorID, err := r.DB.GetCreatorIDForAlias(ctx, request.Alias)
if err != nil {
return err
}
response.UserID = creatorID
return nil
}
// RemoveRoomAlias implements alias.RoomserverInternalAPI
func (r *RoomserverInternalAPI) RemoveRoomAlias(
ctx context.Context,

6
roomserver/internal/api.go
View file

@ -44,7 +44,7 @@ type RoomserverInternalAPI struct {
KeyRing gomatrixserverlib.JSONVerifier
ServerACLs *acls.ServerACLs
fsAPI fsAPI.RoomserverFederationAPI
asAPI asAPI.AppServiceQueryAPI
asAPI asAPI.AppServiceInternalAPI
NATSClient *nats.Conn
JetStream nats.JetStreamContext
Durable string
@ -177,11 +177,11 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
}
}
func (r *RoomserverInternalAPI) SetUserAPI(userAPI userapi.UserInternalAPI) {
func (r *RoomserverInternalAPI) SetUserAPI(userAPI userapi.RoomserverUserAPI) {
r.Leaver.UserAPI = userAPI
}
func (r *RoomserverInternalAPI) SetAppserviceAPI(asAPI asAPI.AppServiceQueryAPI) {
func (r *RoomserverInternalAPI) SetAppserviceAPI(asAPI asAPI.AppServiceInternalAPI) {
r.asAPI = asAPI
}

26
roomserver/internal/input/input_latest_events.go
View file

@ -316,40 +316,30 @@ func (u *latestEventsUpdater) calculateLatest(
// Then let's see if any of the existing forward extremities now
// have entries in the previous events table. If they do then we
// will no longer include them as forward extremities.
existingPrevs := make(map[string]struct{})
for _, l := range existingRefs {
for k, l := range existingRefs {
referenced, err := u.updater.IsReferenced(l.EventReference)
if err != nil {
return false, fmt.Errorf("u.updater.IsReferenced: %w", err)
} else if referenced {
existingPrevs[l.EventID] = struct{}{}
delete(existingRefs, k)
}
}
// Include our new event in the extremities.
newLatest := []types.StateAtEventAndReference{newStateAndRef}
// Start off with our new unreferenced event. We're reusing the backing
// array here rather than allocating a new one.
u.latest = append(u.latest[:0], newStateAndRef)
// Then run through and see if the other extremities are still valid.
// If our new event references them then they are no longer good
// candidates.
// If our new event references any of the existing forward extremities
// then they are no longer forward extremities, so remove them.
for _, prevEventID := range newEvent.PrevEventIDs() {
delete(existingRefs, prevEventID)
}
// Ensure that we don't add any candidate forward extremities from
// the old set that are, themselves, referenced by the old set of
// forward extremities. This shouldn't happen but guards against
// the possibility anyway.
for prevEventID := range existingPrevs {
delete(existingRefs, prevEventID)
}
// Then re-add any old extremities that are still valid after all.
for _, old := range existingRefs {
newLatest = append(newLatest, *old)
u.latest = append(u.latest, *old)
}
u.latest = newLatest
return true, nil
}

12
roomserver/internal/input/input_test.go
View file

@ -10,9 +10,9 @@ import (
"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/setup/process"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
)
@ -21,11 +21,11 @@ var js nats.JetStreamContext
var jc *nats.Conn
func TestMain(m *testing.M) {
var pc *process.ProcessContext
pc, js, jc = jetstream.PrepareForTests()
var b *base.BaseDendrite
b, js, jc = test.Base(nil)
code := m.Run()
pc.ShutdownDendrite()
pc.WaitForComponentsToFinish()
b.ShutdownDendrite()
b.WaitForComponentsToFinish()
os.Exit(code)
}

2
roomserver/internal/perform/perform_leave.go
View file

@ -38,7 +38,7 @@ type Leaver struct {
Cfg *config.RoomServer
DB storage.Database
FSAPI fsAPI.RoomserverFederationAPI
UserAPI userapi.UserInternalAPI
UserAPI userapi.RoomserverUserAPI
Inputer *input.Inputer
}

17
roomserver/inthttp/client.go
View file

@ -91,11 +91,11 @@ func (h *httpRoomserverInternalAPI) SetFederationAPI(fsAPI fsInputAPI.Roomserver
}
// SetAppserviceAPI no-ops in HTTP client mode as there is no chicken/egg scenario
func (h *httpRoomserverInternalAPI) SetAppserviceAPI(asAPI asAPI.AppServiceQueryAPI) {
func (h *httpRoomserverInternalAPI) SetAppserviceAPI(asAPI asAPI.AppServiceInternalAPI) {
}
// SetUserAPI no-ops in HTTP client mode as there is no chicken/egg scenario
func (h *httpRoomserverInternalAPI) SetUserAPI(userAPI userapi.UserInternalAPI) {
func (h *httpRoomserverInternalAPI) SetUserAPI(userAPI userapi.RoomserverUserAPI) {
}
// SetRoomAlias implements RoomserverAliasAPI
@ -137,19 +137,6 @@ func (h *httpRoomserverInternalAPI) GetAliasesForRoomID(
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
// GetCreatorIDForAlias implements RoomserverAliasAPI
func (h *httpRoomserverInternalAPI) GetCreatorIDForAlias(
ctx context.Context,
request *api.GetCreatorIDForAliasRequest,
response *api.GetCreatorIDForAliasResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "GetCreatorIDForAlias")
defer span.Finish()
apiURL := h.roomserverURL + RoomserverGetCreatorIDForAliasPath
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
// RemoveRoomAlias implements RoomserverAliasAPI
func (h *httpRoomserverInternalAPI) RemoveRoomAlias(
ctx context.Context,

14
roomserver/inthttp/server.go
View file

@ -353,20 +353,6 @@ func AddRoutes(r api.RoomserverInternalAPI, internalAPIMux *mux.Router) {
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
internalAPIMux.Handle(
RoomserverGetCreatorIDForAliasPath,
httputil.MakeInternalAPI("GetCreatorIDForAlias", func(req *http.Request) util.JSONResponse {
var request api.GetCreatorIDForAliasRequest
var response api.GetCreatorIDForAliasResponse
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
return util.ErrorResponse(err)
}
if err := r.GetCreatorIDForAlias(req.Context(), &request, &response); err != nil {
return util.ErrorResponse(err)
}
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)
internalAPIMux.Handle(
RoomserverGetAliasesForRoomIDPath,
httputil.MakeInternalAPI("getAliasesForRoomID", func(req *http.Request) util.JSONResponse {

2
roomserver/roomserver.go
View file

@ -50,7 +50,7 @@ func NewInternalAPI(
logrus.WithError(err).Panicf("failed to connect to room server db")
}
js, nc := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
js, nc := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
return internal.NewRoomserverAPI(
base.ProcessContext, cfg, roomserverDB, js, nc,

6
roomserver/storage/postgres/event_json_table.go
View file

@ -59,12 +59,12 @@ type eventJSONStatements struct {
bulkSelectEventJSONStmt *sql.Stmt
}
func createEventJSONTable(db *sql.DB) error {
func CreateEventJSONTable(db *sql.DB) error {
_, err := db.Exec(eventJSONSchema)
return err
}
func prepareEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
func PrepareEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
s := &eventJSONStatements{}
return s, sqlutil.StatementList{
@ -97,9 +97,9 @@ func (s *eventJSONStatements) BulkSelectEventJSON(
// We might get fewer results than NIDs so we adjust the length of the slice before returning it.
results := make([]tables.EventJSONPair, len(eventNIDs))
i := 0
var eventNID int64
for ; rows.Next(); i++ {
result := &results[i]
var eventNID int64
if err := rows.Scan(&eventNID, &result.EventJSON); err != nil {
return nil, err
}

12
roomserver/storage/postgres/event_state_keys_table.go
View file

@ -76,12 +76,12 @@ type eventStateKeyStatements struct {
bulkSelectEventStateKeyStmt *sql.Stmt
}
func createEventStateKeysTable(db *sql.DB) error {
func CreateEventStateKeysTable(db *sql.DB) error {
_, err := db.Exec(eventStateKeysSchema)
return err
}
func prepareEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
func PrepareEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
s := &eventStateKeyStatements{}
return s, sqlutil.StatementList{
@ -123,9 +123,9 @@ func (s *eventStateKeyStatements) BulkSelectEventStateKeyNID(
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventStateKeyNID: rows.close() failed")
result := make(map[string]types.EventStateKeyNID, len(eventStateKeys))
var stateKey string
var stateKeyNID int64
for rows.Next() {
var stateKey string
var stateKeyNID int64
if err := rows.Scan(&stateKey, &stateKeyNID); err != nil {
return nil, err
}
@ -149,9 +149,9 @@ func (s *eventStateKeyStatements) BulkSelectEventStateKey(
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventStateKey: rows.close() failed")
result := make(map[types.EventStateKeyNID]string, len(eventStateKeyNIDs))
var stateKey string
var stateKeyNID int64
for rows.Next() {
var stateKey string
var stateKeyNID int64
if err := rows.Scan(&stateKey, &stateKeyNID); err != nil {
return nil, err
}

8
roomserver/storage/postgres/event_types_table.go
View file

@ -99,12 +99,12 @@ type eventTypeStatements struct {
bulkSelectEventTypeNIDStmt *sql.Stmt
}
func createEventTypesTable(db *sql.DB) error {
func CreateEventTypesTable(db *sql.DB) error {
_, err := db.Exec(eventTypesSchema)
return err
}
func prepareEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
func PrepareEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
s := &eventTypeStatements{}
return s, sqlutil.StatementList{
@ -143,9 +143,9 @@ func (s *eventTypeStatements) BulkSelectEventTypeNID(
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventTypeNID: rows.close() failed")
result := make(map[string]types.EventTypeNID, len(eventTypes))
var eventType string
var eventTypeNID int64
for rows.Next() {
var eventType string
var eventTypeNID int64
if err := rows.Scan(&eventType, &eventTypeNID); err != nil {
return nil, err
}

32
roomserver/storage/postgres/events_table.go
View file

@ -155,12 +155,12 @@ type eventStatements struct {
selectRoomNIDsForEventNIDsStmt *sql.Stmt
}
func createEventsTable(db *sql.DB) error {
func CreateEventsTable(db *sql.DB) error {
_, err := db.Exec(eventsSchema)
return err
}
func prepareEventsTable(db *sql.DB) (tables.Events, error) {
func PrepareEventsTable(db *sql.DB) (tables.Events, error) {
s := &eventStatements{}
return s, sqlutil.StatementList{
@ -380,15 +380,15 @@ func (s *eventStatements) BulkSelectStateAtEventAndReference(
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectStateAtEventAndReference: rows.close() failed")
results := make([]types.StateAtEventAndReference, len(eventNIDs))
i := 0
var (
eventTypeNID int64
eventStateKeyNID int64
eventNID int64
stateSnapshotNID int64
eventID string
eventSHA256 []byte
)
for ; rows.Next(); i++ {
var (
eventTypeNID int64
eventStateKeyNID int64
eventNID int64
stateSnapshotNID int64
eventID string
eventSHA256 []byte
)
if err = rows.Scan(
&eventTypeNID, &eventStateKeyNID, &eventNID, &stateSnapshotNID, &eventID, &eventSHA256,
); err != nil {
@ -446,9 +446,9 @@ func (s *eventStatements) BulkSelectEventID(ctx context.Context, txn *sql.Tx, ev
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventID: rows.close() failed")
results := make(map[types.EventNID]string, len(eventNIDs))
i := 0
var eventNID int64
var eventID string
for ; rows.Next(); i++ {
var eventNID int64
var eventID string
if err = rows.Scan(&eventNID, &eventID); err != nil {
return nil, err
}
@ -491,9 +491,9 @@ func (s *eventStatements) bulkSelectEventNID(ctx context.Context, txn *sql.Tx, e
}
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventNID: rows.close() failed")
results := make(map[string]types.EventNID, len(eventIDs))
var eventID string
var eventNID int64
for rows.Next() {
var eventID string
var eventNID int64
if err = rows.Scan(&eventID, &eventNID); err != nil {
return nil, err
}
@ -522,9 +522,9 @@ func (s *eventStatements) SelectRoomNIDsForEventNIDs(
}
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomNIDsForEventNIDsStmt: rows.close() failed")
result := make(map[types.EventNID]types.RoomNID)
var eventNID types.EventNID
var roomNID types.RoomNID
for rows.Next() {
var eventNID types.EventNID
var roomNID types.RoomNID
if err = rows.Scan(&eventNID, &roomNID); err != nil {
return nil, err
}

16
roomserver/storage/postgres/storage.go
View file

@ -68,16 +68,16 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
}
func (d *Database) create(db *sql.DB) error {
if err := createEventStateKeysTable(db); err != nil {
if err := CreateEventStateKeysTable(db); err != nil {
return err
}
if err := createEventTypesTable(db); err != nil {
if err := CreateEventTypesTable(db); err != nil {
return err
}
if err := createEventJSONTable(db); err != nil {
if err := CreateEventJSONTable(db); err != nil {
return err
}
if err := createEventsTable(db); err != nil {
if err := CreateEventsTable(db); err != nil {
return err
}
if err := createRoomsTable(db); err != nil {
@ -112,19 +112,19 @@ func (d *Database) create(db *sql.DB) error {
}
func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.RoomServerCaches) error {
eventStateKeys, err := prepareEventStateKeysTable(db)
eventStateKeys, err := PrepareEventStateKeysTable(db)
if err != nil {
return err
}
eventTypes, err := prepareEventTypesTable(db)
eventTypes, err := PrepareEventTypesTable(db)
if err != nil {
return err
}
eventJSON, err := prepareEventJSONTable(db)
eventJSON, err := PrepareEventJSONTable(db)
if err != nil {
return err
}
events, err := prepareEventsTable(db)
events, err := PrepareEventsTable(db)
if err != nil {
return err
}

6
roomserver/storage/sqlite3/event_json_table.go
View file

@ -52,12 +52,12 @@ type eventJSONStatements struct {
bulkSelectEventJSONStmt *sql.Stmt
}
func createEventJSONTable(db *sql.DB) error {
func CreateEventJSONTable(db *sql.DB) error {
_, err := db.Exec(eventJSONSchema)
return err
}
func prepareEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
func PrepareEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
s := &eventJSONStatements{
db: db,
}
@ -101,9 +101,9 @@ func (s *eventJSONStatements) BulkSelectEventJSON(
// We might get fewer results than NIDs so we adjust the length of the slice before returning it.
results := make([]tables.EventJSONPair, len(eventNIDs))
i := 0
var eventNID int64
for ; rows.Next(); i++ {
result := &results[i]
var eventNID int64
if err := rows.Scan(&eventNID, &result.EventJSON); err != nil {
return nil, err
}

12
roomserver/storage/sqlite3/event_state_keys_table.go
View file

@ -71,12 +71,12 @@ type eventStateKeyStatements struct {
bulkSelectEventStateKeyStmt *sql.Stmt
}
func createEventStateKeysTable(db *sql.DB) error {
func CreateEventStateKeysTable(db *sql.DB) error {
_, err := db.Exec(eventStateKeysSchema)
return err
}
func prepareEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
func PrepareEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
s := &eventStateKeyStatements{
db: db,
}
@ -128,9 +128,9 @@ func (s *eventStateKeyStatements) BulkSelectEventStateKeyNID(
}
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventStateKeyNID: rows.close() failed")
result := make(map[string]types.EventStateKeyNID, len(eventStateKeys))
var stateKey string
var stateKeyNID int64
for rows.Next() {
var stateKey string
var stateKeyNID int64
if err := rows.Scan(&stateKey, &stateKeyNID); err != nil {
return nil, err
}
@ -159,9 +159,9 @@ func (s *eventStateKeyStatements) BulkSelectEventStateKey(
}
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventStateKey: rows.close() failed")
result := make(map[types.EventStateKeyNID]string, len(eventStateKeyNIDs))
var stateKey string
var stateKeyNID int64
for rows.Next() {
var stateKey string
var stateKeyNID int64
if err := rows.Scan(&stateKey, &stateKeyNID); err != nil {
return nil, err
}

8
roomserver/storage/sqlite3/event_types_table.go
View file

@ -79,12 +79,12 @@ type eventTypeStatements struct {
bulkSelectEventTypeNIDStmt *sql.Stmt
}
func createEventTypesTable(db *sql.DB) error {
func CreateEventTypesTable(db *sql.DB) error {
_, err := db.Exec(eventTypesSchema)
return err
}
func prepareEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
func PrepareEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
s := &eventTypeStatements{
db: db,
}
@ -139,9 +139,9 @@ func (s *eventTypeStatements) BulkSelectEventTypeNID(
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventTypeNID: rows.close() failed")
result := make(map[string]types.EventTypeNID, len(eventTypes))
var eventType string
var eventTypeNID int64
for rows.Next() {
var eventType string
var eventTypeNID int64
if err := rows.Scan(&eventType, &eventTypeNID); err != nil {
return nil, err
}

35
roomserver/storage/sqlite3/events_table.go
View file

@ -68,7 +68,8 @@ const bulkSelectStateEventByIDSQL = "" +
const bulkSelectStateEventByNIDSQL = "" +
"SELECT event_type_nid, event_state_key_nid, event_nid FROM roomserver_events" +
" WHERE event_nid IN ($1)"
// Rest of query is built by BulkSelectStateEventByNID
// Rest of query is built by BulkSelectStateEventByNID
const bulkSelectStateAtEventByIDSQL = "" +
"SELECT event_type_nid, event_state_key_nid, event_nid, state_snapshot_nid, is_rejected FROM roomserver_events" +
@ -126,12 +127,12 @@ type eventStatements struct {
//selectRoomNIDsForEventNIDsStmt *sql.Stmt
}
func createEventsTable(db *sql.DB) error {
func CreateEventsTable(db *sql.DB) error {
_, err := db.Exec(eventsSchema)
return err
}
func prepareEventsTable(db *sql.DB) (tables.Events, error) {
func PrepareEventsTable(db *sql.DB) (tables.Events, error) {
s := &eventStatements{
db: db,
}
@ -404,15 +405,15 @@ func (s *eventStatements) BulkSelectStateAtEventAndReference(
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectStateAtEventAndReference: rows.close() failed")
results := make([]types.StateAtEventAndReference, len(eventNIDs))
i := 0
var (
eventTypeNID int64
eventStateKeyNID int64
eventNID int64
stateSnapshotNID int64
eventID string
eventSHA256 []byte
)
for ; rows.Next(); i++ {
var (
eventTypeNID int64
eventStateKeyNID int64
eventNID int64
stateSnapshotNID int64
eventID string
eventSHA256 []byte
)
if err = rows.Scan(
&eventTypeNID, &eventStateKeyNID, &eventNID, &stateSnapshotNID, &eventID, &eventSHA256,
); err != nil {
@ -491,9 +492,9 @@ func (s *eventStatements) BulkSelectEventID(ctx context.Context, txn *sql.Tx, ev
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventID: rows.close() failed")
results := make(map[types.EventNID]string, len(eventNIDs))
i := 0
var eventNID int64
var eventID string
for ; rows.Next(); i++ {
var eventNID int64
var eventID string
if err = rows.Scan(&eventNID, &eventID); err != nil {
return nil, err
}
@ -545,9 +546,9 @@ func (s *eventStatements) bulkSelectEventNID(ctx context.Context, txn *sql.Tx, e
}
defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectEventNID: rows.close() failed")
results := make(map[string]types.EventNID, len(eventIDs))
var eventID string
var eventNID int64
for rows.Next() {
var eventID string
var eventNID int64
if err = rows.Scan(&eventID, &eventNID); err != nil {
return nil, err
}
@ -595,9 +596,9 @@ func (s *eventStatements) SelectRoomNIDsForEventNIDs(
}
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomNIDsForEventNIDsStmt: rows.close() failed")
result := make(map[types.EventNID]types.RoomNID)
var eventNID types.EventNID
var roomNID types.RoomNID
for rows.Next() {
var eventNID types.EventNID
var roomNID types.RoomNID
if err = rows.Scan(&eventNID, &roomNID); err != nil {
return nil, err
}

16
roomserver/storage/sqlite3/storage.go
View file

@ -77,16 +77,16 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
}
func (d *Database) create(db *sql.DB) error {
if err := createEventStateKeysTable(db); err != nil {
if err := CreateEventStateKeysTable(db); err != nil {
return err
}
if err := createEventTypesTable(db); err != nil {
if err := CreateEventTypesTable(db); err != nil {
return err
}
if err := createEventJSONTable(db); err != nil {
if err := CreateEventJSONTable(db); err != nil {
return err
}
if err := createEventsTable(db); err != nil {
if err := CreateEventsTable(db); err != nil {
return err
}
if err := createRoomsTable(db); err != nil {
@ -121,19 +121,19 @@ func (d *Database) create(db *sql.DB) error {
}
func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.RoomServerCaches) error {
eventStateKeys, err := prepareEventStateKeysTable(db)
eventStateKeys, err := PrepareEventStateKeysTable(db)
if err != nil {
return err
}
eventTypes, err := prepareEventTypesTable(db)
eventTypes, err := PrepareEventTypesTable(db)
if err != nil {
return err
}
eventJSON, err := prepareEventJSONTable(db)
eventJSON, err := PrepareEventJSONTable(db)
if err != nil {
return err
}
events, err := prepareEventsTable(db)
events, err := PrepareEventsTable(db)
if err != nil {
return err
}

95
roomserver/storage/tables/event_json_table_test.go Normal file
View file

@ -0,0 +1,95 @@
package tables_test
import (
"context"
"fmt"
"testing"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/postgres"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/stretchr/testify/assert"
)
func mustCreateEventJSONTable(t *testing.T, dbType test.DBType) (tables.EventJSON, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
db, err := sqlutil.Open(&config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, sqlutil.NewExclusiveWriter())
assert.NoError(t, err)
var tab tables.EventJSON
switch dbType {
case test.DBTypePostgres:
err = postgres.CreateEventJSONTable(db)
assert.NoError(t, err)
tab, err = postgres.PrepareEventJSONTable(db)
case test.DBTypeSQLite:
err = sqlite3.CreateEventJSONTable(db)
assert.NoError(t, err)
tab, err = sqlite3.PrepareEventJSONTable(db)
}
assert.NoError(t, err)
return tab, close
}
func Test_EventJSONTable(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
tab, close := mustCreateEventJSONTable(t, dbType)
defer close()
// create some dummy data
for i := 0; i < 10; i++ {
err := tab.InsertEventJSON(
context.Background(), nil, types.EventNID(i),
[]byte(fmt.Sprintf(`{"value":%d"}`, i)),
)
assert.NoError(t, err)
}
tests := []struct {
name string
args []types.EventNID
wantCount int
}{
{
name: "select subset of existing NIDs",
args: []types.EventNID{1, 2, 3, 4, 5},
wantCount: 5,
},
{
name: "select subset of existing/non-existing NIDs",
args: []types.EventNID{1, 2, 12, 50},
wantCount: 2,
},
{
name: "select single existing NID",
args: []types.EventNID{1},
wantCount: 1,
},
{
name: "select single non-existing NID",
args: []types.EventNID{13},
wantCount: 0,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
// select a subset of the data
values, err := tab.BulkSelectEventJSON(context.Background(), nil, tc.args)
assert.NoError(t, err)
assert.Equal(t, tc.wantCount, len(values))
for i, v := range values {
assert.Equal(t, v.EventNID, types.EventNID(i+1))
assert.Equal(t, []byte(fmt.Sprintf(`{"value":%d"}`, i+1)), v.EventJSON)
}
})
}
})
}

79
roomserver/storage/tables/event_state_keys_table_test.go Normal file
View file

@ -0,0 +1,79 @@
package tables_test
import (
"context"
"fmt"
"testing"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/postgres"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/stretchr/testify/assert"
)
func mustCreateEventStateKeysTable(t *testing.T, dbType test.DBType) (tables.EventStateKeys, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
db, err := sqlutil.Open(&config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, sqlutil.NewExclusiveWriter())
assert.NoError(t, err)
var tab tables.EventStateKeys
switch dbType {
case test.DBTypePostgres:
err = postgres.CreateEventStateKeysTable(db)
assert.NoError(t, err)
tab, err = postgres.PrepareEventStateKeysTable(db)
case test.DBTypeSQLite:
err = sqlite3.CreateEventStateKeysTable(db)
assert.NoError(t, err)
tab, err = sqlite3.PrepareEventStateKeysTable(db)
}
assert.NoError(t, err)
return tab, close
}
func Test_EventStateKeysTable(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
tab, close := mustCreateEventStateKeysTable(t, dbType)
defer close()
ctx := context.Background()
var stateKeyNID, gotEventStateKey types.EventStateKeyNID
var err error
// create some dummy data
for i := 0; i < 10; i++ {
stateKey := fmt.Sprintf("@user%d:localhost", i)
stateKeyNID, err = tab.InsertEventStateKeyNID(ctx, nil, stateKey)
assert.NoError(t, err)
gotEventStateKey, err = tab.SelectEventStateKeyNID(ctx, nil, stateKey)
assert.NoError(t, err)
assert.Equal(t, stateKeyNID, gotEventStateKey)
}
// This should fail, since @user0:localhost already exists
stateKey := fmt.Sprintf("@user%d:localhost", 0)
_, err = tab.InsertEventStateKeyNID(ctx, nil, stateKey)
assert.Error(t, err)
stateKeyNIDsMap, err := tab.BulkSelectEventStateKeyNID(ctx, nil, []string{"@user0:localhost", "@user1:localhost"})
assert.NoError(t, err)
wantStateKeyNIDs := make([]types.EventStateKeyNID, 0, len(stateKeyNIDsMap))
for _, nid := range stateKeyNIDsMap {
wantStateKeyNIDs = append(wantStateKeyNIDs, nid)
}
stateKeyNIDs, err := tab.BulkSelectEventStateKey(ctx, nil, wantStateKeyNIDs)
assert.NoError(t, err)
// verify that BulkSelectEventStateKeyNID and BulkSelectEventStateKey return the same values
for userID, nid := range stateKeyNIDsMap {
if v, ok := stateKeyNIDs[nid]; ok {
assert.Equal(t, v, userID)
} else {
t.Fatalf("unable to find %d in result set", nid)
}
}
})
}

79
roomserver/storage/tables/event_types_table_test.go Normal file
View file

@ -0,0 +1,79 @@
package tables_test
import (
"context"
"fmt"
"testing"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/postgres"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/stretchr/testify/assert"
)
func mustCreateEventTypesTable(t *testing.T, dbType test.DBType) (tables.EventTypes, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
db, err := sqlutil.Open(&config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, sqlutil.NewExclusiveWriter())
assert.NoError(t, err)
var tab tables.EventTypes
switch dbType {
case test.DBTypePostgres:
err = postgres.CreateEventTypesTable(db)
assert.NoError(t, err)
tab, err = postgres.PrepareEventTypesTable(db)
case test.DBTypeSQLite:
err = sqlite3.CreateEventTypesTable(db)
assert.NoError(t, err)
tab, err = sqlite3.PrepareEventTypesTable(db)
}
assert.NoError(t, err)
return tab, close
}
func Test_EventTypesTable(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
tab, close := mustCreateEventTypesTable(t, dbType)
defer close()
ctx := context.Background()
var eventTypeNID, gotEventTypeNID types.EventTypeNID
var err error
// create some dummy data
eventTypeMap := make(map[string]types.EventTypeNID)
for i := 0; i < 10; i++ {
eventType := fmt.Sprintf("dummyEventType%d", i)
eventTypeNID, err = tab.InsertEventTypeNID(ctx, nil, eventType)
assert.NoError(t, err)
eventTypeMap[eventType] = eventTypeNID
gotEventTypeNID, err = tab.SelectEventTypeNID(ctx, nil, eventType)
assert.NoError(t, err)
assert.Equal(t, eventTypeNID, gotEventTypeNID)
}
// This should fail, since the dummyEventType0 already exists
eventType := fmt.Sprintf("dummyEventType%d", 0)
_, err = tab.InsertEventTypeNID(ctx, nil, eventType)
assert.Error(t, err)
// This should return an error, as this eventType does not exist
_, err = tab.SelectEventTypeNID(ctx, nil, "dummyEventType13")
assert.Error(t, err)
eventTypeNIDs, err := tab.BulkSelectEventTypeNID(ctx, nil, []string{"dummyEventType0", "dummyEventType3"})
assert.NoError(t, err)
// verify that BulkSelectEventTypeNID and InsertEventTypeNID return the same values
for eventType, nid := range eventTypeNIDs {
if v, ok := eventTypeMap[eventType]; ok {
assert.Equal(t, v, nid)
} else {
t.Fatalf("unable to find %d in result set", nid)
}
}
})
}

157
roomserver/storage/tables/events_table_test.go Normal file
View file

@ -0,0 +1,157 @@
package tables_test
import (
"context"
"testing"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/postgres"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/gomatrixserverlib"
"github.com/stretchr/testify/assert"
)
func mustCreateEventsTable(t *testing.T, dbType test.DBType) (tables.Events, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
db, err := sqlutil.Open(&config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, sqlutil.NewExclusiveWriter())
assert.NoError(t, err)
var tab tables.Events
switch dbType {
case test.DBTypePostgres:
err = postgres.CreateEventsTable(db)
assert.NoError(t, err)
tab, err = postgres.PrepareEventsTable(db)
case test.DBTypeSQLite:
err = sqlite3.CreateEventsTable(db)
assert.NoError(t, err)
tab, err = sqlite3.PrepareEventsTable(db)
}
assert.NoError(t, err)
return tab, close
}
func Test_EventsTable(t *testing.T) {
alice := test.NewUser()
room := test.NewRoom(t, alice)
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
tab, close := mustCreateEventsTable(t, dbType)
defer close()
// create some dummy data
eventIDs := make([]string, 0, len(room.Events()))
wantStateAtEvent := make([]types.StateAtEvent, 0, len(room.Events()))
wantEventReferences := make([]gomatrixserverlib.EventReference, 0, len(room.Events()))
wantStateAtEventAndRefs := make([]types.StateAtEventAndReference, 0, len(room.Events()))
for _, ev := range room.Events() {
eventNID, snapNID, err := tab.InsertEvent(ctx, nil, 1, 1, 1, ev.EventID(), ev.EventReference().EventSHA256, nil, ev.Depth(), false)
assert.NoError(t, err)
gotEventNID, gotSnapNID, err := tab.SelectEvent(ctx, nil, ev.EventID())
assert.NoError(t, err)
assert.Equal(t, eventNID, gotEventNID)
assert.Equal(t, snapNID, gotSnapNID)
eventID, err := tab.SelectEventID(ctx, nil, eventNID)
assert.NoError(t, err)
assert.Equal(t, eventID, ev.EventID())
// The events shouldn't be sent to output yet
sentToOutput, err := tab.SelectEventSentToOutput(ctx, nil, gotEventNID)
assert.NoError(t, err)
assert.False(t, sentToOutput)
err = tab.UpdateEventSentToOutput(ctx, nil, gotEventNID)
assert.NoError(t, err)
// Now they should be sent to output
sentToOutput, err = tab.SelectEventSentToOutput(ctx, nil, gotEventNID)
assert.NoError(t, err)
assert.True(t, sentToOutput)
eventIDs = append(eventIDs, ev.EventID())
wantEventReferences = append(wantEventReferences, ev.EventReference())
// Set the stateSnapshot to 2 for some events to verify they are returned later
stateSnapshot := 0
if eventNID < 3 {
stateSnapshot = 2
err = tab.UpdateEventState(ctx, nil, eventNID, 2)
assert.NoError(t, err)
}
stateAtEvent := types.StateAtEvent{
Overwrite: false,
BeforeStateSnapshotNID: types.StateSnapshotNID(stateSnapshot),
IsRejected: false,
StateEntry: types.StateEntry{
EventNID: eventNID,
StateKeyTuple: types.StateKeyTuple{
EventTypeNID: 1,
EventStateKeyNID: 1,
},
},
}
wantStateAtEvent = append(wantStateAtEvent, stateAtEvent)
wantStateAtEventAndRefs = append(wantStateAtEventAndRefs, types.StateAtEventAndReference{
StateAtEvent: stateAtEvent,
EventReference: ev.EventReference(),
})
}
stateEvents, err := tab.BulkSelectStateEventByID(ctx, nil, eventIDs)
assert.NoError(t, err)
assert.Equal(t, len(stateEvents), len(eventIDs))
nids := make([]types.EventNID, 0, len(stateEvents))
for _, ev := range stateEvents {
nids = append(nids, ev.EventNID)
}
stateEvents2, err := tab.BulkSelectStateEventByNID(ctx, nil, nids, nil)
assert.NoError(t, err)
// somehow SQLite doesn't return the values ordered as requested by the query
assert.ElementsMatch(t, stateEvents, stateEvents2)
roomNIDs, err := tab.SelectRoomNIDsForEventNIDs(ctx, nil, nids)
assert.NoError(t, err)
// We only inserted one room, so the RoomNID should be the same for all evendNIDs
for _, roomNID := range roomNIDs {
assert.Equal(t, types.RoomNID(1), roomNID)
}
stateAtEvent, err := tab.BulkSelectStateAtEventByID(ctx, nil, eventIDs)
assert.NoError(t, err)
assert.Equal(t, len(eventIDs), len(stateAtEvent))
assert.ElementsMatch(t, wantStateAtEvent, stateAtEvent)
evendNIDMap, err := tab.BulkSelectEventID(ctx, nil, nids)
assert.NoError(t, err)
t.Logf("%+v", evendNIDMap)
assert.Equal(t, len(evendNIDMap), len(nids))
nidMap, err := tab.BulkSelectEventNID(ctx, nil, eventIDs)
assert.NoError(t, err)
// check that we got all expected eventNIDs
for _, eventID := range eventIDs {
_, ok := nidMap[eventID]
assert.True(t, ok)
}
references, err := tab.BulkSelectEventReference(ctx, nil, nids)
assert.NoError(t, err)
assert.Equal(t, wantEventReferences, references)
stateAndRefs, err := tab.BulkSelectStateAtEventAndReference(ctx, nil, nids)
assert.NoError(t, err)
assert.Equal(t, wantStateAtEventAndRefs, stateAndRefs)
// check we get the expected event depth
maxDepth, err := tab.SelectMaxEventDepth(ctx, nil, nids)
assert.NoError(t, err)
assert.Equal(t, int64(len(room.Events())+1), maxDepth)
})
}

8
roomserver/storage/tables/interface.go
View file

@ -10,9 +10,8 @@ import (
)
type EventJSONPair struct {
EventNID types.EventNID
RoomVersion gomatrixserverlib.RoomVersion
EventJSON []byte
EventNID types.EventNID
EventJSON []byte
}
type EventJSON interface {
@ -36,7 +35,8 @@ type EventStateKeys interface {
type Events interface {
InsertEvent(
ctx context.Context, txn *sql.Tx, i types.RoomNID, j types.EventTypeNID, k types.EventStateKeyNID, eventID string,
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventTypeNID types.EventTypeNID,
eventStateKeyNID types.EventStateKeyNID, eventID string,
referenceSHA256 []byte, authEventNIDs []types.EventNID, depth int64, isRejected bool,
) (types.EventNID, types.StateSnapshotNID, error)
SelectEvent(ctx context.Context, txn *sql.Tx, eventID string) (types.EventNID, types.StateSnapshotNID, error)

13
setup/base/base.go
View file

@ -41,6 +41,7 @@ import (
"golang.org/x/net/http2/h2c"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/gorilla/mux"
@ -77,6 +78,7 @@ type BaseDendrite struct {
InternalAPIMux *mux.Router
DendriteAdminMux *mux.Router
SynapseAdminMux *mux.Router
NATS *jetstream.NATSInstance
UseHTTPAPIs bool
apiHttpClient *http.Client
Cfg *config.Dendrite
@ -96,6 +98,7 @@ type BaseDendriteOptions int
const (
NoCacheMetrics BaseDendriteOptions = iota
UseHTTPAPIs
PolylithMode
)
// NewBaseDendrite creates a new instance to be used by a component.
@ -105,17 +108,20 @@ func NewBaseDendrite(cfg *config.Dendrite, componentName string, options ...Base
platformSanityChecks()
useHTTPAPIs := false
cacheMetrics := true
isMonolith := true
for _, opt := range options {
switch opt {
case NoCacheMetrics:
cacheMetrics = false
case UseHTTPAPIs:
useHTTPAPIs = true
case PolylithMode:
isMonolith = false
useHTTPAPIs = true
}
}
configErrors := &config.ConfigErrors{}
isMonolith := componentName == "Monolith" // TODO: better way?
cfg.Verify(configErrors, isMonolith)
if len(*configErrors) > 0 {
for _, err := range *configErrors {
@ -236,6 +242,7 @@ func NewBaseDendrite(cfg *config.Dendrite, componentName string, options ...Base
InternalAPIMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.InternalPathPrefix).Subrouter().UseEncodedPath(),
DendriteAdminMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.DendriteAdminPathPrefix).Subrouter().UseEncodedPath(),
SynapseAdminMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.SynapseAdminPathPrefix).Subrouter().UseEncodedPath(),
NATS: &jetstream.NATSInstance{},
apiHttpClient: &apiClient,
Database: db, // set if monolith with global connection pool only
DatabaseWriter: writer, // set if monolith with global connection pool only
@ -271,8 +278,8 @@ func (b *BaseDendrite) DatabaseConnection(dbProperties *config.DatabaseOptions,
return nil, nil, fmt.Errorf("no database connections configured")
}
// AppserviceHTTPClient returns the AppServiceQueryAPI for hitting the appservice component over HTTP.
func (b *BaseDendrite) AppserviceHTTPClient() appserviceAPI.AppServiceQueryAPI {
// AppserviceHTTPClient returns the AppServiceInternalAPI for hitting the appservice component over HTTP.
func (b *BaseDendrite) AppserviceHTTPClient() appserviceAPI.AppServiceInternalAPI {
a, err := asinthttp.NewAppserviceClient(b.Cfg.AppServiceURL(), b.apiHttpClient)
if err != nil {
logrus.WithError(err).Panic("CreateHTTPAppServiceAPIs failed")

36
setup/jetstream/nats.go
View file

@ -13,31 +13,23 @@ import (
"github.com/sirupsen/logrus"
natsserver "github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
natsclient "github.com/nats-io/nats.go"
)
var natsServer *natsserver.Server
var natsServerMutex sync.Mutex
func PrepareForTests() (*process.ProcessContext, nats.JetStreamContext, *nats.Conn) {
cfg := &config.Dendrite{}
cfg.Defaults(true)
cfg.Global.JetStream.InMemory = true
pc := process.NewProcessContext()
js, jc := Prepare(pc, &cfg.Global.JetStream)
return pc, js, jc
type NATSInstance struct {
*natsserver.Server
sync.Mutex
}
func Prepare(process *process.ProcessContext, cfg *config.JetStream) (natsclient.JetStreamContext, *natsclient.Conn) {
func (s *NATSInstance) Prepare(process *process.ProcessContext, cfg *config.JetStream) (natsclient.JetStreamContext, *natsclient.Conn) {
// check if we need an in-process NATS Server
if len(cfg.Addresses) != 0 {
return setupNATS(process, cfg, nil)
}
natsServerMutex.Lock()
if natsServer == nil {
s.Lock()
if s.Server == nil {
var err error
natsServer, err = natsserver.NewServer(&natsserver.Options{
s.Server, err = natsserver.NewServer(&natsserver.Options{
ServerName: "monolith",
DontListen: true,
JetStream: true,
@ -49,23 +41,23 @@ func Prepare(process *process.ProcessContext, cfg *config.JetStream) (natsclient
if err != nil {
panic(err)
}
natsServer.ConfigureLogger()
s.ConfigureLogger()
go func() {
process.ComponentStarted()
natsServer.Start()
s.Start()
}()
go func() {
<-process.WaitForShutdown()
natsServer.Shutdown()
natsServer.WaitForShutdown()
s.Shutdown()
s.WaitForShutdown()
process.ComponentFinished()
}()
}
natsServerMutex.Unlock()
if !natsServer.ReadyForConnections(time.Second * 10) {
s.Unlock()
if !s.ReadyForConnections(time.Second * 10) {
logrus.Fatalln("NATS did not start in time")
}
nc, err := natsclient.Connect("", natsclient.InProcessServer(natsServer))
nc, err := natsclient.Connect("", natsclient.InProcessServer(s))
if err != nil {
logrus.Fatalln("Failed to create NATS client")
}

2
setup/monolith.go
View file

@ -39,7 +39,7 @@ type Monolith struct {
Client *gomatrixserverlib.Client
FedClient *gomatrixserverlib.FederationClient
AppserviceAPI appserviceAPI.AppServiceQueryAPI
AppserviceAPI appserviceAPI.AppServiceInternalAPI
FederationAPI federationAPI.FederationInternalAPI
RoomserverAPI roomserverAPI.RoomserverInternalAPI
UserAPI userapi.UserInternalAPI

75
syncapi/consumers/roomserver.go
View file

@ -154,41 +154,76 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
ctx context.Context, msg api.OutputNewRoomEvent,
) error {
ev := msg.Event
addsStateEvents := []*gomatrixserverlib.HeaderedEvent{}
foundEventIDs := map[string]bool{}
if len(msg.AddsStateEventIDs) > 0 {
for _, eventID := range msg.AddsStateEventIDs {
foundEventIDs[eventID] = false
// Work out the list of events we need to find out about. Either
// they will be the event supplied in the request, we will find it
// in the sync API database or we'll need to ask the roomserver.
knownEventIDs := make(map[string]bool, len(msg.AddsStateEventIDs))
for _, eventID := range msg.AddsStateEventIDs {
if eventID == ev.EventID() {
knownEventIDs[eventID] = true
addsStateEvents = append(addsStateEvents, ev)
} else {
knownEventIDs[eventID] = false
}
foundEvents, err := s.db.Events(ctx, msg.AddsStateEventIDs)
}
// Work out which events we want to look up in the sync API database.
// At this stage the only event that should be excluded is the event
// supplied in the request, if it appears in the adds_state_event_ids.
missingEventIDs := make([]string, 0, len(msg.AddsStateEventIDs))
for eventID, known := range knownEventIDs {
if !known {
missingEventIDs = append(missingEventIDs, eventID)
}
}
// Look the events up in the database. If we know them, add them into
// the set of adds state events.
if len(missingEventIDs) > 0 {
alreadyKnown, err := s.db.Events(ctx, msg.AddsStateEventIDs)
if err != nil {
return fmt.Errorf("s.db.Events: %w", err)
}
for _, event := range foundEvents {
foundEventIDs[event.EventID()] = true
for _, knownEvent := range alreadyKnown {
knownEventIDs[knownEvent.EventID()] = true
addsStateEvents = append(addsStateEvents, knownEvent)
}
}
// Now work out if there are any remaining events we don't know. For
// these we will need to ask the roomserver for help.
missingEventIDs = missingEventIDs[:0]
for eventID, known := range knownEventIDs {
if !known {
missingEventIDs = append(missingEventIDs, eventID)
}
}
// Ask the roomserver and add in the rest of the results into the set.
// Finally, work out if there are any more events missing.
if len(missingEventIDs) > 0 {
eventsReq := &api.QueryEventsByIDRequest{
EventIDs: missingEventIDs,
}
eventsReq := &api.QueryEventsByIDRequest{}
eventsRes := &api.QueryEventsByIDResponse{}
for eventID, found := range foundEventIDs {
if !found {
eventsReq.EventIDs = append(eventsReq.EventIDs, eventID)
}
}
if err = s.rsAPI.QueryEventsByID(ctx, eventsReq, eventsRes); err != nil {
if err := s.rsAPI.QueryEventsByID(ctx, eventsReq, eventsRes); err != nil {
return fmt.Errorf("s.rsAPI.QueryEventsByID: %w", err)
}
for _, event := range eventsRes.Events {
eventID := event.EventID()
foundEvents = append(foundEvents, event)
foundEventIDs[eventID] = true
addsStateEvents = append(addsStateEvents, event)
knownEventIDs[event.EventID()] = true
}
for eventID, found := range foundEventIDs {
// This should never happen because this would imply that the
// roomserver has sent us adds_state_event_ids for events that it
// also doesn't know about, but let's just be sure.
for eventID, found := range knownEventIDs {
if !found {
return fmt.Errorf("event %s is missing", eventID)
}
}
addsStateEvents = foundEvents
}
ev, err := s.updateStateEvent(ev)

4
syncapi/routing/context.go
View file

@ -45,7 +45,7 @@ func Context(
rsAPI roomserver.SyncRoomserverAPI,
syncDB storage.Database,
roomID, eventID string,
lazyLoadCache *caching.LazyLoadCache,
lazyLoadCache caching.LazyLoadCache,
) util.JSONResponse {
filter, err := parseRoomEventFilter(req)
if err != nil {
@ -155,7 +155,7 @@ func applyLazyLoadMembers(
filter *gomatrixserverlib.RoomEventFilter,
eventsAfter, eventsBefore []gomatrixserverlib.ClientEvent,
state []*gomatrixserverlib.HeaderedEvent,
lazyLoadCache *caching.LazyLoadCache,
lazyLoadCache caching.LazyLoadCache,
) []*gomatrixserverlib.HeaderedEvent {
if filter == nil || !filter.LazyLoadMembers {
return state

2
syncapi/routing/messages.go
View file

@ -63,7 +63,7 @@ func OnIncomingMessagesRequest(
rsAPI api.SyncRoomserverAPI,
cfg *config.SyncAPI,
srp *sync.RequestPool,
lazyLoadCache *caching.LazyLoadCache,
lazyLoadCache caching.LazyLoadCache,
) util.JSONResponse {
var err error

2
syncapi/routing/routing.go
View file

@ -39,7 +39,7 @@ func Setup(
userAPI userapi.SyncUserAPI,
rsAPI api.SyncRoomserverAPI,
cfg *config.SyncAPI,
lazyLoadCache *caching.LazyLoadCache,
lazyLoadCache caching.LazyLoadCache,
) {
v3mux := csMux.PathPrefix("/{apiversion:(?:r0|v3)}/").Subrouter()

2
syncapi/streams/stream_pdu.go
View file

@ -32,7 +32,7 @@ type PDUStreamProvider struct {
tasks chan func()
workers atomic.Int32
// userID+deviceID -> lazy loading cache
lazyLoadCache *caching.LazyLoadCache
lazyLoadCache caching.LazyLoadCache
rsAPI roomserverAPI.SyncRoomserverAPI
}

2
syncapi/streams/streams.go
View file

@ -27,7 +27,7 @@ type Streams struct {
func NewSyncStreamProviders(
d storage.Database, userAPI userapi.SyncUserAPI,
rsAPI rsapi.SyncRoomserverAPI, keyAPI keyapi.SyncKeyAPI,
eduCache *caching.EDUCache, lazyLoadCache *caching.LazyLoadCache, notifier *notifier.Notifier,
eduCache *caching.EDUCache, lazyLoadCache caching.LazyLoadCache, notifier *notifier.Notifier,
) *Streams {
streams := &Streams{
PDUStreamProvider: &PDUStreamProvider{

10
syncapi/syncapi.go
View file

@ -45,7 +45,7 @@ func AddPublicRoutes(
) {
cfg := &base.Cfg.SyncAPI
js, natsClient := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
js, natsClient := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
syncDB, err := storage.NewSyncServerDatasource(base, &cfg.Database)
if err != nil {
@ -53,12 +53,8 @@ func AddPublicRoutes(
}
eduCache := caching.NewTypingCache()
lazyLoadCache, err := caching.NewLazyLoadCache()
if err != nil {
logrus.WithError(err).Panicf("failed to create lazy loading cache")
}
notifier := notifier.NewNotifier()
streams := streams.NewSyncStreamProviders(syncDB, userAPI, rsAPI, keyAPI, eduCache, lazyLoadCache, notifier)
streams := streams.NewSyncStreamProviders(syncDB, userAPI, rsAPI, keyAPI, eduCache, base.Caches, notifier)
notifier.SetCurrentPosition(streams.Latest(context.Background()))
if err = notifier.Load(context.Background(), syncDB); err != nil {
logrus.WithError(err).Panicf("failed to load notifier ")
@ -146,6 +142,6 @@ func AddPublicRoutes(
routing.Setup(
base.PublicClientAPIMux, requestPool, syncDB, userAPI,
rsAPI, cfg, lazyLoadCache,
rsAPI, cfg, base.Caches,
)
}

18
test/base.go Normal file
View file

@ -0,0 +1,18 @@
package test
import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/nats-io/nats.go"
)
func Base(cfg *config.Dendrite) (*base.BaseDendrite, nats.JetStreamContext, *nats.Conn) {
if cfg == nil {
cfg = &config.Dendrite{}
cfg.Defaults(true)
}
cfg.Global.JetStream.InMemory = true
base := base.NewBaseDendrite(cfg, "Tests")
js, jc := base.NATS.Prepare(base.ProcessContext, &cfg.Global.JetStream)
return base, js, jc
}

11
userapi/api/api.go
View file

@ -31,6 +31,8 @@ type UserInternalAPI interface {
ClientUserAPI
MediaUserAPI
FederationUserAPI
RoomserverUserAPI
KeyserverUserAPI
QuerySearchProfilesAPI // used by p2p demos
}
@ -41,6 +43,15 @@ type AppserviceUserAPI interface {
PerformDeviceCreation(ctx context.Context, req *PerformDeviceCreationRequest, res *PerformDeviceCreationResponse) error
}
type KeyserverUserAPI interface {
QueryDevices(ctx context.Context, req *QueryDevicesRequest, res *QueryDevicesResponse) error
QueryDeviceInfos(ctx context.Context, req *QueryDeviceInfosRequest, res *QueryDeviceInfosResponse) error
}
type RoomserverUserAPI interface {
QueryAccountData(ctx context.Context, req *QueryAccountDataRequest, res *QueryAccountDataResponse) error
}
// api functions required by the media api
type MediaUserAPI interface {
QueryAcccessTokenAPI

2
userapi/userapi.go
View file

@ -47,7 +47,7 @@ func NewInternalAPI(
appServices []config.ApplicationService, keyAPI keyapi.UserKeyAPI,
rsAPI rsapi.UserRoomserverAPI, pgClient pushgateway.Client,
) api.UserInternalAPI {
js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
db, err := storage.NewUserAPIDatabase(
base,