mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-29 01:33:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Initial federation sender -> federation API refactoring

Browse source
This commit is contained in:
Neil Alexander 2021-11-18 16:30:05 +00:00
parent 403498a85b
commit 0276aedeef
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
91 changed files with 468 additions and 496 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
build/docker/config/dendrite-config.yaml
View file

@ -187,7 +187,7 @@ federation_sender:
listen: http://0.0.0.0:7775
connect: http://federation_sender:7775
database:
connection_string: postgresql://dendrite:itsasecret@postgres/dendrite_federationsender?sslmode=disable
connection_string: postgresql://dendrite:itsasecret@postgres/dendrite_federationapi?sslmode=disable
max_open_conns: 10
max_idle_conns: 2
conn_max_lifetime: -1

2
build/docker/docker-compose.polylith.yml
View file

@ -58,7 +58,7 @@ services:
federation_sender:
hostname: federation_sender
image: matrixdotorg/dendrite-polylith:latest
command: federationsender
command: federationapi
volumes:
- ./config:/etc/dendrite
networks:

2
build/docker/postgres/create_db.sh
View file

@ -1,5 +1,5 @@
#!/bin/sh
for db in userapi_accounts userapi_devices mediaapi syncapi roomserver signingkeyserver keyserver federationsender appservice naffka; do
for db in userapi_accounts userapi_devices mediaapi syncapi roomserver signingkeyserver keyserver federationapi appservice naffka; do
createdb -U dendrite -O dendrite dendrite_$db
done

8
build/gobind-pinecone/monolith.go
View file

@ -24,8 +24,8 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/roomserver"
@ -272,7 +272,7 @@ func (m *DendriteMonolith) Start() {
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-roomserver.db", m.StorageDirectory, prefix))
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-signingkeyserver.db", m.StorageDirectory, prefix))
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-keyserver.db", m.StorageDirectory, prefix))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-federationsender.db", m.StorageDirectory, prefix))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-federationapi.db", m.StorageDirectory, prefix))
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-appservice.db", m.StorageDirectory, prefix))
cfg.MediaAPI.BasePath = config.Path(fmt.Sprintf("%s/media", m.CacheDirectory))
cfg.MediaAPI.AbsBasePath = config.Path(fmt.Sprintf("%s/media", m.CacheDirectory))
@ -294,7 +294,7 @@ func (m *DendriteMonolith) Start() {
base, keyRing,
)
fsAPI := federationsender.NewInternalAPI(
fsAPI := federationapi.NewInternalAPI(
base, federation, rsAPI, keyRing, true,
)

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

@ -15,8 +15,8 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/yggrooms"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/roomserver"
@ -94,7 +94,7 @@ func (m *DendriteMonolith) Start() {
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-roomserver.db", m.StorageDirectory))
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-signingkeyserver.db", m.StorageDirectory))
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-keyserver.db", m.StorageDirectory))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-federationsender.db", m.StorageDirectory))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-federationapi.db", m.StorageDirectory))
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-appservice.db", m.StorageDirectory))
cfg.MediaAPI.BasePath = config.Path(fmt.Sprintf("%s/tmp", m.StorageDirectory))
cfg.MediaAPI.AbsBasePath = config.Path(fmt.Sprintf("%s/tmp", m.StorageDirectory))
@ -115,7 +115,7 @@ func (m *DendriteMonolith) Start() {
base, keyRing,
)
fsAPI := federationsender.NewInternalAPI(
fsAPI := federationapi.NewInternalAPI(
base, federation, rsAPI, keyRing, true,
)

4
clientapi/clientapi.go
View file

@ -21,7 +21,7 @@ import (
"github.com/matrix-org/dendrite/clientapi/producers"
"github.com/matrix-org/dendrite/clientapi/routing"
eduServerAPI "github.com/matrix-org/dendrite/eduserver/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/transactions"
keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
@ -43,7 +43,7 @@ func AddPublicRoutes(
eduInputAPI eduServerAPI.EDUServerInputAPI,
asAPI appserviceAPI.AppServiceQueryAPI,
transactionsCache *transactions.Cache,
fsAPI federationSenderAPI.FederationSenderInternalAPI,
fsAPI federationSenderAPI.FederationInternalAPI,
userAPI userapi.UserInternalAPI,
keyAPI keyserverAPI.KeyInternalAPI,
extRoomsProvider api.ExtraPublicRoomsProvider,

4
clientapi/routing/directory.go
View file

@ -20,7 +20,7 @@ import (
"github.com/matrix-org/dendrite/clientapi/httputil"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
@ -47,7 +47,7 @@ func DirectoryRoom(
federation *gomatrixserverlib.FederationClient,
cfg *config.ClientAPI,
rsAPI roomserverAPI.RoomserverInternalAPI,
fedSenderAPI federationSenderAPI.FederationSenderInternalAPI,
fedSenderAPI federationSenderAPI.FederationInternalAPI,
) util.JSONResponse {
_, domain, err := gomatrixserverlib.SplitID('#', roomAlias)
if err != nil {

4
clientapi/routing/routing.go
View file

@ -27,7 +27,7 @@ import (
"github.com/matrix-org/dendrite/clientapi/jsonerror"
"github.com/matrix-org/dendrite/clientapi/producers"
eduServerAPI "github.com/matrix-org/dendrite/eduserver/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/transactions"
keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
@ -56,7 +56,7 @@ func Setup(
federation *gomatrixserverlib.FederationClient,
syncProducer *producers.SyncAPIProducer,
transactionsCache *transactions.Cache,
federationSender federationSenderAPI.FederationSenderInternalAPI,
federationSender federationSenderAPI.FederationInternalAPI,
keyAPI keyserverAPI.KeyInternalAPI,
extRoomsProvider api.ExtraPublicRoomsProvider,
mscCfg *config.MSCs,

6
cmd/dendrite-demo-libp2p/main.go
View file

@ -30,7 +30,7 @@ import (
"github.com/matrix-org/dendrite/appservice"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/embed"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/roomserver"
@ -132,7 +132,7 @@ func main() {
cfg.SyncAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-syncapi.db", *instanceName))
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", *instanceName))
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-signingkeyserver.db", *instanceName))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationapi.db", *instanceName))
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
cfg.Global.Kafka.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-naffka.db", *instanceName))
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-e2ekey.db", *instanceName))
@ -167,7 +167,7 @@ func main() {
)
asAPI := appservice.NewInternalAPI(&base.Base, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
fsAPI := federationsender.NewInternalAPI(
fsAPI := federationapi.NewInternalAPI(
&base.Base, federation, rsAPI, keyRing, true,
)
rsAPI.SetFederationSenderAPI(fsAPI)

8
cmd/dendrite-demo-pinecone/main.go
View file

@ -37,8 +37,8 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
@ -153,7 +153,7 @@ func main() {
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", *instanceName))
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-signingkeyserver.db", *instanceName))
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-keyserver.db", *instanceName))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationapi.db", *instanceName))
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
cfg.Global.Kafka.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-naffka.db", *instanceName))
cfg.MSCs.MSCs = []string{"msc2836", "msc2946"}
@ -174,7 +174,7 @@ func main() {
base, keyRing,
)
rsAPI := rsComponent
fsAPI := federationsender.NewInternalAPI(
fsAPI := federationapi.NewInternalAPI(
base, federation, rsAPI, keyRing, true,
)

6
cmd/dendrite-demo-pinecone/rooms/rooms.go
View file

@ -19,7 +19,7 @@ import (
"sync"
"time"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@ -30,14 +30,14 @@ import (
type PineconeRoomProvider struct {
r *pineconeRouter.Router
s *pineconeSessions.Sessions
fedSender api.FederationSenderInternalAPI
fedSender api.FederationInternalAPI
fedClient *gomatrixserverlib.FederationClient
}
func NewPineconeRoomProvider(
r *pineconeRouter.Router,
s *pineconeSessions.Sessions,
fedSender api.FederationSenderInternalAPI,
fedSender api.FederationInternalAPI,
fedClient *gomatrixserverlib.FederationClient,
) *PineconeRoomProvider {
p := &PineconeRoomProvider{

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

@ -31,8 +31,8 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/yggrooms"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
@ -84,7 +84,7 @@ func main() {
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", *instanceName))
cfg.SigningKeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-signingkeyserver.db", *instanceName))
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-keyserver.db", *instanceName))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
cfg.FederationSender.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationapi.db", *instanceName))
cfg.AppServiceAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
cfg.Global.Kafka.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-naffka.db", *instanceName))
cfg.MSCs.MSCs = []string{"msc2836"}
@ -117,7 +117,7 @@ func main() {
asAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
fsAPI := federationsender.NewInternalAPI(
fsAPI := federationapi.NewInternalAPI(
base, federation, rsAPI, keyRing, true,
)

6
cmd/dendrite-demo-yggdrasil/yggrooms/yggrooms.go
View file

@ -20,19 +20,19 @@ import (
"time"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/yggconn"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
type YggdrasilRoomProvider struct {
node *yggconn.Node
fedSender api.FederationSenderInternalAPI
fedSender api.FederationInternalAPI
fedClient *gomatrixserverlib.FederationClient
}
func NewYggdrasilRoomProvider(
node *yggconn.Node, fedSender api.FederationSenderInternalAPI, fedClient *gomatrixserverlib.FederationClient,
node *yggconn.Node, fedSender api.FederationInternalAPI, fedClient *gomatrixserverlib.FederationClient,
) *YggdrasilRoomProvider {
p := &YggdrasilRoomProvider{
node: node,

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

@ -21,7 +21,7 @@ import (
"github.com/matrix-org/dendrite/appservice"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/roomserver/api"
@ -101,11 +101,11 @@ func main() {
}
}
fsAPI := federationsender.NewInternalAPI(
fsAPI := federationapi.NewInternalAPI(
base, federation, rsAPI, keyRing, false,
)
if base.UseHTTPAPIs {
federationsender.AddInternalRoutes(base.InternalAPIMux, fsAPI)
federationapi.AddInternalRoutes(base.InternalAPIMux, fsAPI)
fsAPI = base.FederationSenderHTTPClient()
}
// The underlying roomserver implementation needs to be able to call the fedsender.

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

@ -44,7 +44,7 @@ func main() {
"clientapi": personalities.ClientAPI,
"eduserver": personalities.EDUServer,
"federationapi": personalities.FederationAPI,
"federationsender": personalities.FederationSender,
"federationapi": personalities.FederationSender,
"keyserver": personalities.KeyServer,
"mediaapi": personalities.MediaAPI,
"roomserver": personalities.RoomServer,

6
cmd/dendrite-polylith-multi/personalities/federationsender.go
View file

@ -15,7 +15,7 @@
package personalities
import (
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/config"
)
@ -27,10 +27,10 @@ func FederationSender(base *setup.BaseDendrite, cfg *config.Dendrite) {
keyRing := serverKeyAPI.KeyRing()
rsAPI := base.RoomserverHTTPClient()
fsAPI := federationsender.NewInternalAPI(
fsAPI := federationapi.NewInternalAPI(
base, federation, rsAPI, keyRing, false,
)
federationsender.AddInternalRoutes(base.InternalAPIMux, fsAPI)
federationapi.AddInternalRoutes(base.InternalAPIMux, fsAPI)
base.SetupAndServeHTTP(
base.Cfg.FederationSender.InternalAPI.Listen, // internal listener

4
cmd/dendritejs-pinecone/main.go
View file

@ -33,7 +33,7 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/roomserver"
@ -198,7 +198,7 @@ func startup() {
base, userAPI, rsAPI,
)
rsAPI.SetAppserviceAPI(asQuery)
fedSenderAPI := federationsender.NewInternalAPI(base, federation, rsAPI, keyRing, true)
fedSenderAPI := federationapi.NewInternalAPI(base, federation, rsAPI, keyRing, true)
rsAPI.SetFederationSenderAPI(fedSenderAPI)
monolith := setup.Monolith{

4
cmd/dendritejs/main.go
View file

@ -26,7 +26,7 @@ import (
"github.com/matrix-org/dendrite/appservice"
"github.com/matrix-org/dendrite/eduserver"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/federationsender"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/roomserver"
@ -211,7 +211,7 @@ func main() {
base, userAPI, rsAPI,
)
rsAPI.SetAppserviceAPI(asQuery)
fedSenderAPI := federationsender.NewInternalAPI(base, federation, rsAPI, &keyRing, true)
fedSenderAPI := federationapi.NewInternalAPI(base, federation, rsAPI, &keyRing, true)
rsAPI.SetFederationSenderAPI(fedSenderAPI)
p2pPublicRoomProvider := NewLibP2PPublicRoomsProvider(node, fedSenderAPI, federation)

6
cmd/dendritejs/publicrooms.go
View file

@ -22,7 +22,7 @@ import (
"sync"
"time"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi/api"
go_http_js_libp2p "github.com/matrix-org/go-http-js-libp2p"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@ -31,12 +31,12 @@ import (
type libp2pPublicRoomsProvider struct {
node *go_http_js_libp2p.P2pLocalNode
providers []go_http_js_libp2p.PeerInfo
fedSender api.FederationSenderInternalAPI
fedSender api.FederationInternalAPI
fedClient *gomatrixserverlib.FederationClient
}
func NewLibP2PPublicRoomsProvider(
node *go_http_js_libp2p.P2pLocalNode, fedSender api.FederationSenderInternalAPI, fedClient *gomatrixserverlib.FederationClient,
node *go_http_js_libp2p.P2pLocalNode, fedSender api.FederationInternalAPI, fedClient *gomatrixserverlib.FederationClient,
) *libp2pPublicRoomsProvider {
p := &libp2pPublicRoomsProvider{
node: node,

2
cmd/goose/main.go
View file

@ -20,7 +20,7 @@ import (
const (
AppService = "appservice"
FederationSender = "federationsender"
FederationSender = "federationapi"
KeyServer = "keyserver"
MediaAPI = "mediaapi"
RoomServer = "roomserver"

6
federationsender/api/api.go → federationapi/api/api.go
View file

@ -5,7 +5,7 @@ import (
"fmt"
"time"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
)
@ -36,8 +36,8 @@ func (e *FederationClientError) Error() string {
return fmt.Sprintf("%s - (retry_after=%s, blacklisted=%v)", e.Err, e.RetryAfter.String(), e.Blacklisted)
}
// FederationSenderInternalAPI is used to query information from the federation sender.
type FederationSenderInternalAPI interface {
// FederationInternalAPI is used to query information from the federation sender.
type FederationInternalAPI interface {
FederationClient
QueryServerKeys(ctx context.Context, request *QueryServerKeysRequest, response *QueryServerKeysResponse) error

4
federationsender/consumers/eduserver.go → federationapi/consumers/eduserver.go
View file

@ -21,8 +21,8 @@ import (
"github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/federationsender/queue"
"github.com/matrix-org/dendrite/federationsender/storage"
"github.com/matrix-org/dendrite/federationapi/queue"
"github.com/matrix-org/dendrite/federationapi/storage"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/process"

6
federationsender/consumers/keychange.go → federationapi/consumers/keychange.go
View file

@ -21,8 +21,8 @@ import (
"github.com/Shopify/sarama"
eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/federationsender/queue"
"github.com/matrix-org/dendrite/federationsender/storage"
"github.com/matrix-org/dendrite/federationapi/queue"
"github.com/matrix-org/dendrite/federationapi/storage"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/keyserver/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
@ -53,7 +53,7 @@ func NewKeyChangeConsumer(
c := &KeyChangeConsumer{
consumer: &internal.ContinualConsumer{
Process: process,
ComponentName: "federationsender/keychange",
ComponentName: "federationapi/keychange",
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputKeyChangeEvent)),
Consumer: kafkaConsumer,
PartitionStore: store,

10
federationsender/consumers/roomserver.go → federationapi/consumers/roomserver.go
View file

@ -20,9 +20,9 @@ import (
"fmt"
"github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/federationsender/queue"
"github.com/matrix-org/dendrite/federationsender/storage"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/queue"
"github.com/matrix-org/dendrite/federationapi/storage"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
@ -51,7 +51,7 @@ func NewOutputRoomEventConsumer(
) *OutputRoomEventConsumer {
consumer := internal.ContinualConsumer{
Process: process,
ComponentName: "federationsender/roomserver",
ComponentName: "federationapi/roomserver",
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputRoomEvent)),
Consumer: kafkaConsumer,
PartitionStore: store,
@ -133,7 +133,7 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
}
// processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any)
// causing the federationsender to start sending messages to the peeking server
// causing the federationapi to start sending messages to the peeking server
func (s *OutputRoomEventConsumer) processInboundPeek(orp api.OutputNewInboundPeek) error {
// FIXME: there's a race here - we should start /sending new peeked events

0
federationsender/consumers/roomserver_test.go → federationapi/consumers/roomserver_test.go
View file

75
federationapi/federationapi.go
View file

@ -17,12 +17,21 @@ package federationapi
import (
"github.com/gorilla/mux"
eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/federationapi/api"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/consumers"
"github.com/matrix-org/dendrite/federationapi/internal"
"github.com/matrix-org/dendrite/federationapi/queue"
"github.com/matrix-org/dendrite/federationapi/statistics"
"github.com/matrix-org/dendrite/federationapi/storage"
keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/kafka"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/federationapi/routing"
"github.com/matrix-org/gomatrixserverlib"
@ -36,7 +45,7 @@ func AddPublicRoutes(
federation *gomatrixserverlib.FederationClient,
keyRing gomatrixserverlib.JSONVerifier,
rsAPI roomserverAPI.RoomserverInternalAPI,
federationSenderAPI federationSenderAPI.FederationSenderInternalAPI,
federationSenderAPI federationSenderAPI.FederationInternalAPI,
eduAPI eduserverAPI.EDUServerInputAPI,
keyAPI keyserverAPI.KeyInternalAPI,
mscCfg *config.MSCs,
@ -49,3 +58,65 @@ func AddPublicRoutes(
servers,
)
}
// NewInternalAPI returns a concerete implementation of the internal API. Callers
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
func NewInternalAPI(
base *setup.BaseDendrite,
federation *gomatrixserverlib.FederationClient,
rsAPI roomserverAPI.RoomserverInternalAPI,
keyRing *gomatrixserverlib.KeyRing,
resetBlacklist bool,
) api.FederationInternalAPI {
cfg := &base.Cfg.FederationSender
federationSenderDB, err := storage.NewDatabase(&cfg.Database, base.Caches)
if err != nil {
logrus.WithError(err).Panic("failed to connect to federation sender db")
}
if resetBlacklist {
_ = federationSenderDB.RemoveAllServersFromBlacklist()
}
stats := &statistics.Statistics{
DB: federationSenderDB,
FailuresUntilBlacklist: cfg.FederationMaxRetries,
}
consumer, _ := kafka.SetupConsumerProducer(&cfg.Matrix.Kafka)
queues := queue.NewOutgoingQueues(
federationSenderDB, base.ProcessContext,
cfg.Matrix.DisableFederation,
cfg.Matrix.ServerName, federation, rsAPI, stats,
&queue.SigningInfo{
KeyID: cfg.Matrix.KeyID,
PrivateKey: cfg.Matrix.PrivateKey,
ServerName: cfg.Matrix.ServerName,
},
)
rsConsumer := consumers.NewOutputRoomEventConsumer(
base.ProcessContext, cfg, consumer, queues,
federationSenderDB, rsAPI,
)
if err = rsConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start room server consumer")
}
tsConsumer := consumers.NewOutputEDUConsumer(
base.ProcessContext, cfg, consumer, queues, federationSenderDB,
)
if err := tsConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start typing server consumer")
}
keyConsumer := consumers.NewKeyChangeConsumer(
base.ProcessContext, &base.Cfg.KeyServer, consumer, queues, federationSenderDB, rsAPI,
)
if err := keyConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start key server consumer")
}
return internal.NewFederationInternalAPI(federationSenderDB, cfg, rsAPI, federation, keyRing, stats, queues)
}

42
federationsender/internal/api.go → federationapi/internal/api.go
View file

@ -5,18 +5,18 @@ import (
"sync"
"time"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationsender/queue"
"github.com/matrix-org/dendrite/federationsender/statistics"
"github.com/matrix-org/dendrite/federationsender/storage"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/queue"
"github.com/matrix-org/dendrite/federationapi/statistics"
"github.com/matrix-org/dendrite/federationapi/storage"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
)
// FederationSenderInternalAPI is an implementation of api.FederationSenderInternalAPI
type FederationSenderInternalAPI struct {
// FederationInternalAPI is an implementation of api.FederationInternalAPI
type FederationInternalAPI struct {
db storage.Database
cfg *config.FederationSender
statistics *statistics.Statistics
@ -27,15 +27,15 @@ type FederationSenderInternalAPI struct {
joins sync.Map // joins currently in progress
}
func NewFederationSenderInternalAPI(
func NewFederationInternalAPI(
db storage.Database, cfg *config.FederationSender,
rsAPI roomserverAPI.RoomserverInternalAPI,
federation *gomatrixserverlib.FederationClient,
keyRing *gomatrixserverlib.KeyRing,
statistics *statistics.Statistics,
queues *queue.OutgoingQueues,
) *FederationSenderInternalAPI {
return &FederationSenderInternalAPI{
) *FederationInternalAPI {
return &FederationInternalAPI{
db: db,
cfg: cfg,
rsAPI: rsAPI,
@ -46,7 +46,7 @@ func NewFederationSenderInternalAPI(
}
}
func (a *FederationSenderInternalAPI) isBlacklistedOrBackingOff(s gomatrixserverlib.ServerName) (*statistics.ServerStatistics, error) {
func (a *FederationInternalAPI) isBlacklistedOrBackingOff(s gomatrixserverlib.ServerName) (*statistics.ServerStatistics, error) {
stats := a.statistics.ForServer(s)
until, blacklisted := stats.BackoffInfo()
if blacklisted {
@ -81,7 +81,7 @@ func failBlacklistableError(err error, stats *statistics.ServerStatistics) (unti
return
}
func (a *FederationSenderInternalAPI) doRequest(
func (a *FederationInternalAPI) doRequest(
s gomatrixserverlib.ServerName, request func() (interface{}, error),
) (interface{}, error) {
stats, err := a.isBlacklistedOrBackingOff(s)
@ -106,7 +106,7 @@ func (a *FederationSenderInternalAPI) doRequest(
return res, nil
}
func (a *FederationSenderInternalAPI) GetUserDevices(
func (a *FederationInternalAPI) GetUserDevices(
ctx context.Context, s gomatrixserverlib.ServerName, userID string,
) (gomatrixserverlib.RespUserDevices, error) {
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
@ -120,7 +120,7 @@ func (a *FederationSenderInternalAPI) GetUserDevices(
return ires.(gomatrixserverlib.RespUserDevices), nil
}
func (a *FederationSenderInternalAPI) ClaimKeys(
func (a *FederationInternalAPI) ClaimKeys(
ctx context.Context, s gomatrixserverlib.ServerName, oneTimeKeys map[string]map[string]string,
) (gomatrixserverlib.RespClaimKeys, error) {
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
@ -134,7 +134,7 @@ func (a *FederationSenderInternalAPI) ClaimKeys(
return ires.(gomatrixserverlib.RespClaimKeys), nil
}
func (a *FederationSenderInternalAPI) QueryKeys(
func (a *FederationInternalAPI) QueryKeys(
ctx context.Context, s gomatrixserverlib.ServerName, keys map[string][]string,
) (gomatrixserverlib.RespQueryKeys, error) {
ires, err := a.doRequest(s, func() (interface{}, error) {
@ -146,7 +146,7 @@ func (a *FederationSenderInternalAPI) QueryKeys(
return ires.(gomatrixserverlib.RespQueryKeys), nil
}
func (a *FederationSenderInternalAPI) Backfill(
func (a *FederationInternalAPI) Backfill(
ctx context.Context, s gomatrixserverlib.ServerName, roomID string, limit int, eventIDs []string,
) (res gomatrixserverlib.Transaction, err error) {
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
@ -160,7 +160,7 @@ func (a *FederationSenderInternalAPI) Backfill(
return ires.(gomatrixserverlib.Transaction), nil
}
func (a *FederationSenderInternalAPI) LookupState(
func (a *FederationInternalAPI) LookupState(
ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string, roomVersion gomatrixserverlib.RoomVersion,
) (res gomatrixserverlib.RespState, err error) {
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
@ -174,7 +174,7 @@ func (a *FederationSenderInternalAPI) LookupState(
return ires.(gomatrixserverlib.RespState), nil
}
func (a *FederationSenderInternalAPI) LookupStateIDs(
func (a *FederationInternalAPI) LookupStateIDs(
ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string,
) (res gomatrixserverlib.RespStateIDs, err error) {
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
@ -188,7 +188,7 @@ func (a *FederationSenderInternalAPI) LookupStateIDs(
return ires.(gomatrixserverlib.RespStateIDs), nil
}
func (a *FederationSenderInternalAPI) GetEvent(
func (a *FederationInternalAPI) GetEvent(
ctx context.Context, s gomatrixserverlib.ServerName, eventID string,
) (res gomatrixserverlib.Transaction, err error) {
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
@ -202,7 +202,7 @@ func (a *FederationSenderInternalAPI) GetEvent(
return ires.(gomatrixserverlib.Transaction), nil
}
func (a *FederationSenderInternalAPI) LookupServerKeys(
func (a *FederationInternalAPI) LookupServerKeys(
ctx context.Context, s gomatrixserverlib.ServerName, keyRequests map[gomatrixserverlib.PublicKeyLookupRequest]gomatrixserverlib.Timestamp,
) ([]gomatrixserverlib.ServerKeys, error) {
ctx, cancel := context.WithTimeout(ctx, time.Minute)
@ -216,7 +216,7 @@ func (a *FederationSenderInternalAPI) LookupServerKeys(
return ires.([]gomatrixserverlib.ServerKeys), nil
}
func (a *FederationSenderInternalAPI) MSC2836EventRelationships(
func (a *FederationInternalAPI) MSC2836EventRelationships(
ctx context.Context, s gomatrixserverlib.ServerName, r gomatrixserverlib.MSC2836EventRelationshipsRequest,
roomVersion gomatrixserverlib.RoomVersion,
) (res gomatrixserverlib.MSC2836EventRelationshipsResponse, err error) {
@ -231,7 +231,7 @@ func (a *FederationSenderInternalAPI) MSC2836EventRelationships(
return ires.(gomatrixserverlib.MSC2836EventRelationshipsResponse), nil
}
func (a *FederationSenderInternalAPI) MSC2946Spaces(
func (a *FederationInternalAPI) MSC2946Spaces(
ctx context.Context, s gomatrixserverlib.ServerName, roomID string, r gomatrixserverlib.MSC2946SpacesRequest,
) (res gomatrixserverlib.MSC2946SpacesResponse, err error) {
ctx, cancel := context.WithTimeout(ctx, time.Minute)

34
federationsender/internal/perform.go → federationapi/internal/perform.go
View file

@ -7,7 +7,7 @@ import (
"fmt"
"time"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/version"
"github.com/matrix-org/gomatrix"
@ -16,8 +16,8 @@ import (
"github.com/sirupsen/logrus"
)
// PerformLeaveRequest implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformDirectoryLookup(
// PerformLeaveRequest implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformDirectoryLookup(
ctx context.Context,
request *api.PerformDirectoryLookupRequest,
response *api.PerformDirectoryLookupResponse,
@ -42,8 +42,8 @@ type federatedJoin struct {
RoomID string
}
// PerformJoin implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformJoin(
// PerformJoin implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformJoin(
ctx context.Context,
request *api.PerformJoinRequest,
response *api.PerformJoinResponse,
@ -132,7 +132,7 @@ func (r *FederationSenderInternalAPI) PerformJoin(
)
}
func (r *FederationSenderInternalAPI) performJoinUsingServer(
func (r *FederationInternalAPI) performJoinUsingServer(
ctx context.Context,
roomID, userID string,
content map[string]interface{},
@ -263,8 +263,8 @@ func (r *FederationSenderInternalAPI) performJoinUsingServer(
return nil
}
// PerformOutboundPeekRequest implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformOutboundPeek(
// PerformOutboundPeekRequest implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformOutboundPeek(
ctx context.Context,
request *api.PerformOutboundPeekRequest,
response *api.PerformOutboundPeekResponse,
@ -344,7 +344,7 @@ func (r *FederationSenderInternalAPI) PerformOutboundPeek(
return lastErr
}
func (r *FederationSenderInternalAPI) performOutboundPeekUsingServer(
func (r *FederationInternalAPI) performOutboundPeekUsingServer(
ctx context.Context,
roomID string,
serverName gomatrixserverlib.ServerName,
@ -438,8 +438,8 @@ func (r *FederationSenderInternalAPI) performOutboundPeekUsingServer(
return nil
}
// PerformLeaveRequest implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformLeave(
// PerformLeaveRequest implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformLeave(
ctx context.Context,
request *api.PerformLeaveRequest,
response *api.PerformLeaveResponse,
@ -528,8 +528,8 @@ func (r *FederationSenderInternalAPI) PerformLeave(
)
}
// PerformLeaveRequest implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformInvite(
// PerformLeaveRequest implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformInvite(
ctx context.Context,
request *api.PerformInviteRequest,
response *api.PerformInviteResponse,
@ -565,8 +565,8 @@ func (r *FederationSenderInternalAPI) PerformInvite(
return nil
}
// PerformServersAlive implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformServersAlive(
// PerformServersAlive implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformServersAlive(
ctx context.Context,
request *api.PerformServersAliveRequest,
response *api.PerformServersAliveResponse,
@ -579,8 +579,8 @@ func (r *FederationSenderInternalAPI) PerformServersAlive(
return nil
}
// PerformServersAlive implements api.FederationSenderInternalAPI
func (r *FederationSenderInternalAPI) PerformBroadcastEDU(
// PerformServersAlive implements api.FederationInternalAPI
func (r *FederationInternalAPI) PerformBroadcastEDU(
ctx context.Context,
request *api.PerformBroadcastEDURequest,
response *api.PerformBroadcastEDUResponse,

12
federationsender/internal/query.go → federationapi/internal/query.go
View file

@ -5,13 +5,13 @@ import (
"fmt"
"time"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
// QueryJoinedHostServerNamesInRoom implements api.FederationSenderInternalAPI
func (f *FederationSenderInternalAPI) QueryJoinedHostServerNamesInRoom(
// QueryJoinedHostServerNamesInRoom implements api.FederationInternalAPI
func (f *FederationInternalAPI) QueryJoinedHostServerNamesInRoom(
ctx context.Context,
request *api.QueryJoinedHostServerNamesInRoomRequest,
response *api.QueryJoinedHostServerNamesInRoomResponse,
@ -25,7 +25,7 @@ func (f *FederationSenderInternalAPI) QueryJoinedHostServerNamesInRoom(
return
}
func (a *FederationSenderInternalAPI) fetchServerKeysDirectly(ctx context.Context, serverName gomatrixserverlib.ServerName) (*gomatrixserverlib.ServerKeys, error) {
func (a *FederationInternalAPI) fetchServerKeysDirectly(ctx context.Context, serverName gomatrixserverlib.ServerName) (*gomatrixserverlib.ServerKeys, error) {
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
defer cancel()
ires, err := a.doRequest(serverName, func() (interface{}, error) {
@ -38,7 +38,7 @@ func (a *FederationSenderInternalAPI) fetchServerKeysDirectly(ctx context.Contex
return &sks, nil
}
func (a *FederationSenderInternalAPI) fetchServerKeysFromCache(
func (a *FederationInternalAPI) fetchServerKeysFromCache(
ctx context.Context, req *api.QueryServerKeysRequest,
) ([]gomatrixserverlib.ServerKeys, error) {
var results []gomatrixserverlib.ServerKeys
@ -64,7 +64,7 @@ func (a *FederationSenderInternalAPI) fetchServerKeysFromCache(
return results, nil
}
func (a *FederationSenderInternalAPI) QueryServerKeys(
func (a *FederationInternalAPI) QueryServerKeys(
ctx context.Context, req *api.QueryServerKeysRequest, res *api.QueryServerKeysResponse,
) error {
// attempt to satisfy the entire request from the cache first

90
federationsender/inthttp/client.go → federationapi/inthttp/client.go
View file

@ -5,7 +5,7 @@ import (
"errors"
"net/http"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
@ -14,45 +14,45 @@ import (
// HTTP paths for the internal HTTP API
const (
FederationSenderQueryJoinedHostServerNamesInRoomPath = "/federationsender/queryJoinedHostServerNamesInRoom"
FederationSenderQueryServerKeysPath = "/federationsender/queryServerKeys"
FederationSenderQueryJoinedHostServerNamesInRoomPath = "/federationapi/queryJoinedHostServerNamesInRoom"
FederationSenderQueryServerKeysPath = "/federationapi/queryServerKeys"
FederationSenderPerformDirectoryLookupRequestPath = "/federationsender/performDirectoryLookup"
FederationSenderPerformJoinRequestPath = "/federationsender/performJoinRequest"
FederationSenderPerformLeaveRequestPath = "/federationsender/performLeaveRequest"
FederationSenderPerformInviteRequestPath = "/federationsender/performInviteRequest"
FederationSenderPerformOutboundPeekRequestPath = "/federationsender/performOutboundPeekRequest"
FederationSenderPerformServersAlivePath = "/federationsender/performServersAlive"
FederationSenderPerformBroadcastEDUPath = "/federationsender/performBroadcastEDU"
FederationSenderPerformDirectoryLookupRequestPath = "/federationapi/performDirectoryLookup"
FederationSenderPerformJoinRequestPath = "/federationapi/performJoinRequest"
FederationSenderPerformLeaveRequestPath = "/federationapi/performLeaveRequest"
FederationSenderPerformInviteRequestPath = "/federationapi/performInviteRequest"
FederationSenderPerformOutboundPeekRequestPath = "/federationapi/performOutboundPeekRequest"
FederationSenderPerformServersAlivePath = "/federationapi/performServersAlive"
FederationSenderPerformBroadcastEDUPath = "/federationapi/performBroadcastEDU"
FederationSenderGetUserDevicesPath = "/federationsender/client/getUserDevices"
FederationSenderClaimKeysPath = "/federationsender/client/claimKeys"
FederationSenderQueryKeysPath = "/federationsender/client/queryKeys"
FederationSenderBackfillPath = "/federationsender/client/backfill"
FederationSenderLookupStatePath = "/federationsender/client/lookupState"
FederationSenderLookupStateIDsPath = "/federationsender/client/lookupStateIDs"
FederationSenderGetEventPath = "/federationsender/client/getEvent"
FederationSenderLookupServerKeysPath = "/federationsender/client/lookupServerKeys"
FederationSenderEventRelationshipsPath = "/federationsender/client/msc2836eventRelationships"
FederationSenderSpacesSummaryPath = "/federationsender/client/msc2946spacesSummary"
FederationSenderGetUserDevicesPath = "/federationapi/client/getUserDevices"
FederationSenderClaimKeysPath = "/federationapi/client/claimKeys"
FederationSenderQueryKeysPath = "/federationapi/client/queryKeys"
FederationSenderBackfillPath = "/federationapi/client/backfill"
FederationSenderLookupStatePath = "/federationapi/client/lookupState"
FederationSenderLookupStateIDsPath = "/federationapi/client/lookupStateIDs"
FederationSenderGetEventPath = "/federationapi/client/getEvent"
FederationSenderLookupServerKeysPath = "/federationapi/client/lookupServerKeys"
FederationSenderEventRelationshipsPath = "/federationapi/client/msc2836eventRelationships"
FederationSenderSpacesSummaryPath = "/federationapi/client/msc2946spacesSummary"
)
// NewFederationSenderClient creates a FederationSenderInternalAPI implemented by talking to a HTTP POST API.
// NewFederationSenderClient creates a FederationInternalAPI implemented by talking to a HTTP POST API.
// If httpClient is nil an error is returned
func NewFederationSenderClient(federationSenderURL string, httpClient *http.Client) (api.FederationSenderInternalAPI, error) {
func NewFederationSenderClient(federationSenderURL string, httpClient *http.Client) (api.FederationInternalAPI, error) {
if httpClient == nil {
return nil, errors.New("NewFederationSenderInternalAPIHTTP: httpClient is <nil>")
return nil, errors.New("NewFederationInternalAPIHTTP: httpClient is <nil>")
}
return &httpFederationSenderInternalAPI{federationSenderURL, httpClient}, nil
return &httpFederationInternalAPI{federationSenderURL, httpClient}, nil
}
type httpFederationSenderInternalAPI struct {
type httpFederationInternalAPI struct {
federationSenderURL string
httpClient *http.Client
}
// Handle an instruction to make_leave & send_leave with a remote server.
func (h *httpFederationSenderInternalAPI) PerformLeave(
func (h *httpFederationInternalAPI) PerformLeave(
ctx context.Context,
request *api.PerformLeaveRequest,
response *api.PerformLeaveResponse,
@ -65,7 +65,7 @@ func (h *httpFederationSenderInternalAPI) PerformLeave(
}
// Handle sending an invite to a remote server.
func (h *httpFederationSenderInternalAPI) PerformInvite(
func (h *httpFederationInternalAPI) PerformInvite(
ctx context.Context,
request *api.PerformInviteRequest,
response *api.PerformInviteResponse,
@ -78,7 +78,7 @@ func (h *httpFederationSenderInternalAPI) PerformInvite(
}
// Handle starting a peek on a remote server.
func (h *httpFederationSenderInternalAPI) PerformOutboundPeek(
func (h *httpFederationInternalAPI) PerformOutboundPeek(
ctx context.Context,
request *api.PerformOutboundPeekRequest,
response *api.PerformOutboundPeekResponse,
@ -90,7 +90,7 @@ func (h *httpFederationSenderInternalAPI) PerformOutboundPeek(
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
func (h *httpFederationSenderInternalAPI) PerformServersAlive(
func (h *httpFederationInternalAPI) PerformServersAlive(
ctx context.Context,
request *api.PerformServersAliveRequest,
response *api.PerformServersAliveResponse,
@ -102,8 +102,8 @@ func (h *httpFederationSenderInternalAPI) PerformServersAlive(
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
}
// QueryJoinedHostServerNamesInRoom implements FederationSenderInternalAPI
func (h *httpFederationSenderInternalAPI) QueryJoinedHostServerNamesInRoom(
// QueryJoinedHostServerNamesInRoom implements FederationInternalAPI
func (h *httpFederationInternalAPI) QueryJoinedHostServerNamesInRoom(
ctx context.Context,
request *api.QueryJoinedHostServerNamesInRoomRequest,
response *api.QueryJoinedHostServerNamesInRoomResponse,
@ -116,7 +116,7 @@ func (h *httpFederationSenderInternalAPI) QueryJoinedHostServerNamesInRoom(
}
// Handle an instruction to make_join & send_join with a remote server.
func (h *httpFederationSenderInternalAPI) PerformJoin(
func (h *httpFederationInternalAPI) PerformJoin(
ctx context.Context,
request *api.PerformJoinRequest,
response *api.PerformJoinResponse,
@ -136,7 +136,7 @@ func (h *httpFederationSenderInternalAPI) PerformJoin(
}
// Handle an instruction to make_join & send_join with a remote server.
func (h *httpFederationSenderInternalAPI) PerformDirectoryLookup(
func (h *httpFederationInternalAPI) PerformDirectoryLookup(
ctx context.Context,
request *api.PerformDirectoryLookupRequest,
response *api.PerformDirectoryLookupResponse,
@ -149,7 +149,7 @@ func (h *httpFederationSenderInternalAPI) PerformDirectoryLookup(
}
// Handle an instruction to broadcast an EDU to all servers in rooms we are joined to.
func (h *httpFederationSenderInternalAPI) PerformBroadcastEDU(
func (h *httpFederationInternalAPI) PerformBroadcastEDU(
ctx context.Context,
request *api.PerformBroadcastEDURequest,
response *api.PerformBroadcastEDUResponse,
@ -168,7 +168,7 @@ type getUserDevices struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) GetUserDevices(
func (h *httpFederationInternalAPI) GetUserDevices(
ctx context.Context, s gomatrixserverlib.ServerName, userID string,
) (gomatrixserverlib.RespUserDevices, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "GetUserDevices")
@ -198,7 +198,7 @@ type claimKeys struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) ClaimKeys(
func (h *httpFederationInternalAPI) ClaimKeys(
ctx context.Context, s gomatrixserverlib.ServerName, oneTimeKeys map[string]map[string]string,
) (gomatrixserverlib.RespClaimKeys, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "ClaimKeys")
@ -228,7 +228,7 @@ type queryKeys struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) QueryKeys(
func (h *httpFederationInternalAPI) QueryKeys(
ctx context.Context, s gomatrixserverlib.ServerName, keys map[string][]string,
) (gomatrixserverlib.RespQueryKeys, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "QueryKeys")
@ -260,7 +260,7 @@ type backfill struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) Backfill(
func (h *httpFederationInternalAPI) Backfill(
ctx context.Context, s gomatrixserverlib.ServerName, roomID string, limit int, eventIDs []string,
) (gomatrixserverlib.Transaction, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "Backfill")
@ -293,7 +293,7 @@ type lookupState struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) LookupState(
func (h *httpFederationInternalAPI) LookupState(
ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string, roomVersion gomatrixserverlib.RoomVersion,
) (gomatrixserverlib.RespState, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "LookupState")
@ -325,7 +325,7 @@ type lookupStateIDs struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) LookupStateIDs(
func (h *httpFederationInternalAPI) LookupStateIDs(
ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string,
) (gomatrixserverlib.RespStateIDs, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "LookupStateIDs")
@ -355,7 +355,7 @@ type getEvent struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) GetEvent(
func (h *httpFederationInternalAPI) GetEvent(
ctx context.Context, s gomatrixserverlib.ServerName, eventID string,
) (gomatrixserverlib.Transaction, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "GetEvent")
@ -377,7 +377,7 @@ func (h *httpFederationSenderInternalAPI) GetEvent(
return *response.Res, nil
}
func (h *httpFederationSenderInternalAPI) QueryServerKeys(
func (h *httpFederationInternalAPI) QueryServerKeys(
ctx context.Context, req *api.QueryServerKeysRequest, res *api.QueryServerKeysResponse,
) error {
span, ctx := opentracing.StartSpanFromContext(ctx, "QueryServerKeys")
@ -394,7 +394,7 @@ type lookupServerKeys struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) LookupServerKeys(
func (h *httpFederationInternalAPI) LookupServerKeys(
ctx context.Context, s gomatrixserverlib.ServerName, keyRequests map[gomatrixserverlib.PublicKeyLookupRequest]gomatrixserverlib.Timestamp,
) ([]gomatrixserverlib.ServerKeys, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "LookupServerKeys")
@ -424,7 +424,7 @@ type eventRelationships struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) MSC2836EventRelationships(
func (h *httpFederationInternalAPI) MSC2836EventRelationships(
ctx context.Context, s gomatrixserverlib.ServerName, r gomatrixserverlib.MSC2836EventRelationshipsRequest,
roomVersion gomatrixserverlib.RoomVersion,
) (res gomatrixserverlib.MSC2836EventRelationshipsResponse, err error) {
@ -456,7 +456,7 @@ type spacesReq struct {
Err *api.FederationClientError
}
func (h *httpFederationSenderInternalAPI) MSC2946Spaces(
func (h *httpFederationInternalAPI) MSC2946Spaces(
ctx context.Context, dst gomatrixserverlib.ServerName, roomID string, r gomatrixserverlib.MSC2946SpacesRequest,
) (res gomatrixserverlib.MSC2946SpacesResponse, err error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "MSC2946Spaces")

6
federationsender/inthttp/server.go → federationapi/inthttp/server.go
View file

@ -5,14 +5,14 @@ import (
"net/http"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/util"
)
// AddRoutes adds the FederationSenderInternalAPI handlers to the http.ServeMux.
// AddRoutes adds the FederationInternalAPI handlers to the http.ServeMux.
// nolint:gocyclo
func AddRoutes(intAPI api.FederationSenderInternalAPI, internalAPIMux *mux.Router) {
func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationSenderQueryJoinedHostServerNamesInRoomPath,
httputil.MakeInternalAPI("QueryJoinedHostServerNamesInRoom", func(req *http.Request) util.JSONResponse {

10
federationsender/queue/destinationqueue.go → federationapi/queue/destinationqueue.go
View file

@ -21,9 +21,9 @@ import (
"sync"
"time"
"github.com/matrix-org/dendrite/federationsender/statistics"
"github.com/matrix-org/dendrite/federationsender/storage"
"github.com/matrix-org/dendrite/federationsender/storage/shared"
"github.com/matrix-org/dendrite/federationapi/statistics"
"github.com/matrix-org/dendrite/federationapi/storage"
"github.com/matrix-org/dendrite/federationapi/storage/shared"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrix"
@ -81,7 +81,7 @@ func (oq *destinationQueue) sendEvent(event *gomatrixserverlib.HeaderedEvent, re
context.TODO(),
"", // TODO: remove this, as we don't need to persist the transaction ID
oq.destination, // the destination server name
receipt, // NIDs from federationsender_queue_json table
receipt, // NIDs from federationapi_queue_json table
); err != nil {
logrus.WithError(err).Errorf("failed to associate PDU %q with destination %q", event.EventID(), oq.destination)
return
@ -124,7 +124,7 @@ func (oq *destinationQueue) sendEDU(event *gomatrixserverlib.EDU, receipt *share
if err := oq.db.AssociateEDUWithDestination(
context.TODO(),
oq.destination, // the destination server name
receipt, // NIDs from federationsender_queue_json table
receipt, // NIDs from federationapi_queue_json table
); err != nil {
logrus.WithError(err).Errorf("failed to associate EDU with destination %q", oq.destination)
return

12
federationsender/queue/queue.go → federationapi/queue/queue.go
View file

@ -22,9 +22,9 @@ import (
"sync"
"time"
"github.com/matrix-org/dendrite/federationsender/statistics"
"github.com/matrix-org/dendrite/federationsender/storage"
"github.com/matrix-org/dendrite/federationsender/storage/shared"
"github.com/matrix-org/dendrite/federationapi/statistics"
"github.com/matrix-org/dendrite/federationapi/storage"
"github.com/matrix-org/dendrite/federationapi/storage/shared"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
@ -58,7 +58,7 @@ func init() {
var destinationQueueTotal = prometheus.NewGauge(
prometheus.GaugeOpts{
Namespace: "dendrite",
Subsystem: "federationsender",
Subsystem: "federationapi",
Name: "destination_queues_total",
},
)
@ -66,7 +66,7 @@ var destinationQueueTotal = prometheus.NewGauge(
var destinationQueueRunning = prometheus.NewGauge(
prometheus.GaugeOpts{
Namespace: "dendrite",
Subsystem: "federationsender",
Subsystem: "federationapi",
Name: "destination_queues_running",
},
)
@ -74,7 +74,7 @@ var destinationQueueRunning = prometheus.NewGauge(
var destinationQueueBackingOff = prometheus.NewGauge(
prometheus.GaugeOpts{
Namespace: "dendrite",
Subsystem: "federationsender",
Subsystem: "federationapi",
Name: "destination_queues_backing_off",
},
)

4
federationapi/routing/keys.go
View file

@ -21,7 +21,7 @@ import (
"github.com/matrix-org/dendrite/clientapi/httputil"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
@ -179,7 +179,7 @@ func localKeys(cfg *config.FederationAPI, validUntil time.Time) (*gomatrixserver
func NotaryKeys(
httpReq *http.Request, cfg *config.FederationAPI,
fsAPI federationSenderAPI.FederationSenderInternalAPI,
fsAPI federationSenderAPI.FederationInternalAPI,
req *gomatrixserverlib.PublicKeyNotaryLookupRequest,
) util.JSONResponse {
if req == nil {

2
federationapi/routing/peek.go
View file

@ -33,7 +33,7 @@ func Peek(
roomID, peekID string,
remoteVersions []gomatrixserverlib.RoomVersion,
) util.JSONResponse {
// TODO: check if we're just refreshing an existing peek by querying the federationsender
// TODO: check if we're just refreshing an existing peek by querying the federationapi
verReq := api.QueryRoomVersionForRoomRequest{RoomID: roomID}
verRes := api.QueryRoomVersionForRoomResponse{}

4
federationapi/routing/query.go
View file

@ -19,7 +19,7 @@ import (
"net/http"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrix"
@ -33,7 +33,7 @@ func RoomAliasToID(
federation *gomatrixserverlib.FederationClient,
cfg *config.FederationAPI,
rsAPI roomserverAPI.RoomserverInternalAPI,
senderAPI federationSenderAPI.FederationSenderInternalAPI,
senderAPI federationSenderAPI.FederationInternalAPI,
) util.JSONResponse {
roomAlias := httpReq.FormValue("room_alias")
if roomAlias == "" {

4
federationapi/routing/routing.go
View file

@ -21,7 +21,7 @@ import (
"github.com/matrix-org/dendrite/clientapi/jsonerror"
eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
@ -46,7 +46,7 @@ func Setup(
cfg *config.FederationAPI,
rsAPI roomserverAPI.RoomserverInternalAPI,
eduAPI eduserverAPI.EDUServerInputAPI,
fsAPI federationSenderAPI.FederationSenderInternalAPI,
fsAPI federationSenderAPI.FederationInternalAPI,
keys gomatrixserverlib.JSONVerifier,
federation *gomatrixserverlib.FederationClient,
userAPI userapi.UserInternalAPI,

2
federationsender/statistics/statistics.go → federationapi/statistics/statistics.go
View file

@ -5,7 +5,7 @@ import (
"sync"
"time"
"github.com/matrix-org/dendrite/federationsender/storage"
"github.com/matrix-org/dendrite/federationapi/storage"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
"go.uber.org/atomic"

0
federationsender/statistics/statistics_test.go → federationapi/statistics/statistics_test.go
View file

4
federationsender/storage/interface.go → federationapi/storage/interface.go
View file

@ -17,8 +17,8 @@ package storage
import (
"context"
"github.com/matrix-org/dendrite/federationsender/storage/shared"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/storage/shared"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/gomatrixserverlib"
)

10
federationsender/storage/postgres/blacklist_table.go → federationapi/storage/postgres/blacklist_table.go
View file

@ -23,7 +23,7 @@ import (
)
const blacklistSchema = `
CREATE TABLE IF NOT EXISTS federationsender_blacklist (
CREATE TABLE IF NOT EXISTS federationapi_blacklist (
-- The blacklisted server name
server_name TEXT NOT NULL,
UNIQUE (server_name)
@ -31,17 +31,17 @@ CREATE TABLE IF NOT EXISTS federationsender_blacklist (
`
const insertBlacklistSQL = "" +
"INSERT INTO federationsender_blacklist (server_name) VALUES ($1)" +
"INSERT INTO federationapi_blacklist (server_name) VALUES ($1)" +
" ON CONFLICT DO NOTHING"
const selectBlacklistSQL = "" +
"SELECT server_name FROM federationsender_blacklist WHERE server_name = $1"
"SELECT server_name FROM federationapi_blacklist WHERE server_name = $1"
const deleteBlacklistSQL = "" +
"DELETE FROM federationsender_blacklist WHERE server_name = $1"
"DELETE FROM federationapi_blacklist WHERE server_name = $1"
const deleteAllBlacklistSQL = "" +
"TRUNCATE federationsender_blacklist"
"TRUNCATE federationapi_blacklist"
type blacklistStatements struct {
db *sql.DB

2
federationsender/storage/postgres/deltas/2021020411080000_rooms.go → federationapi/storage/postgres/deltas/2021020411080000_rooms.go
View file

@ -32,7 +32,7 @@ func LoadRemoveRoomsTable(m *sqlutil.Migrations) {
func UpRemoveRoomsTable(tx *sql.Tx) error {
_, err := tx.Exec(`
DROP TABLE IF EXISTS federationsender_rooms;
DROP TABLE IF EXISTS federationapi_rooms;
`)
if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err)

16
federationsender/storage/postgres/inbound_peeks_table.go → federationapi/storage/postgres/inbound_peeks_table.go
View file

@ -19,14 +19,14 @@ import (
"database/sql"
"time"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
)
const inboundPeeksSchema = `
CREATE TABLE IF NOT EXISTS federationsender_inbound_peeks (
CREATE TABLE IF NOT EXISTS federationapi_inbound_peeks (
room_id TEXT NOT NULL,
server_name TEXT NOT NULL,
peek_id TEXT NOT NULL,
@ -38,22 +38,22 @@ CREATE TABLE IF NOT EXISTS federationsender_inbound_peeks (
`
const insertInboundPeekSQL = "" +
"INSERT INTO federationsender_inbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
"INSERT INTO federationapi_inbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
const selectInboundPeekSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_inbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_inbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
const selectInboundPeeksSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_inbound_peeks WHERE room_id = $1"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_inbound_peeks WHERE room_id = $1"
const renewInboundPeekSQL = "" +
"UPDATE federationsender_inbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
"UPDATE federationapi_inbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
const deleteInboundPeekSQL = "" +
"DELETE FROM federationsender_inbound_peeks WHERE room_id = $1 and server_name = $2"
"DELETE FROM federationapi_inbound_peeks WHERE room_id = $1 and server_name = $2"
const deleteInboundPeeksSQL = "" +
"DELETE FROM federationsender_inbound_peeks WHERE room_id = $1"
"DELETE FROM federationapi_inbound_peeks WHERE room_id = $1"
type inboundPeeksStatements struct {
db *sql.DB

20
federationsender/storage/postgres/joined_hosts_table.go → federationapi/storage/postgres/joined_hosts_table.go
View file

@ -20,7 +20,7 @@ import (
"database/sql"
"github.com/lib/pq"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
@ -30,7 +30,7 @@ const joinedHostsSchema = `
-- The joined_hosts table stores a list of m.room.member event ids in the
-- current state for each room where the membership is "join".
-- There will be an entry for every user that is joined to the room.
CREATE TABLE IF NOT EXISTS federationsender_joined_hosts (
CREATE TABLE IF NOT EXISTS federationapi_joined_hosts (
-- The string ID of the room.
room_id TEXT NOT NULL,
-- The event ID of the m.room.member join event.
@ -40,31 +40,31 @@ CREATE TABLE IF NOT EXISTS federationsender_joined_hosts (
);
CREATE UNIQUE INDEX IF NOT EXISTS federatonsender_joined_hosts_event_id_idx
ON federationsender_joined_hosts (event_id);
ON federationapi_joined_hosts (event_id);
CREATE INDEX IF NOT EXISTS federatonsender_joined_hosts_room_id_idx
ON federationsender_joined_hosts (room_id)
ON federationapi_joined_hosts (room_id)
`
const insertJoinedHostsSQL = "" +
"INSERT INTO federationsender_joined_hosts (room_id, event_id, server_name)" +
"INSERT INTO federationapi_joined_hosts (room_id, event_id, server_name)" +
" VALUES ($1, $2, $3) ON CONFLICT DO NOTHING"
const deleteJoinedHostsSQL = "" +
"DELETE FROM federationsender_joined_hosts WHERE event_id = ANY($1)"
"DELETE FROM federationapi_joined_hosts WHERE event_id = ANY($1)"
const deleteJoinedHostsForRoomSQL = "" +
"DELETE FROM federationsender_joined_hosts WHERE room_id = $1"
"DELETE FROM federationapi_joined_hosts WHERE room_id = $1"
const selectJoinedHostsSQL = "" +
"SELECT event_id, server_name FROM federationsender_joined_hosts" +
"SELECT event_id, server_name FROM federationapi_joined_hosts" +
" WHERE room_id = $1"
const selectAllJoinedHostsSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_joined_hosts"
"SELECT DISTINCT server_name FROM federationapi_joined_hosts"
const selectJoinedHostsForRoomsSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_joined_hosts WHERE room_id = ANY($1)"
"SELECT DISTINCT server_name FROM federationapi_joined_hosts WHERE room_id = ANY($1)"
type joinedHostsStatements struct {
db *sql.DB

10
federationsender/storage/postgres/notary_server_keys_json_table.go → federationapi/storage/postgres/notary_server_keys_json_table.go
View file

@ -18,14 +18,14 @@ import (
"context"
"database/sql"
"github.com/matrix-org/dendrite/federationsender/storage/tables"
"github.com/matrix-org/dendrite/federationapi/storage/tables"
"github.com/matrix-org/gomatrixserverlib"
)
const notaryServerKeysJSONSchema = `
CREATE SEQUENCE IF NOT EXISTS federationsender_notary_server_keys_json_pkey;
CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_json (
notary_id BIGINT PRIMARY KEY NOT NULL DEFAULT nextval('federationsender_notary_server_keys_json_pkey'),
CREATE SEQUENCE IF NOT EXISTS federationapi_notary_server_keys_json_pkey;
CREATE TABLE IF NOT EXISTS federationapi_notary_server_keys_json (
notary_id BIGINT PRIMARY KEY NOT NULL DEFAULT nextval('federationapi_notary_server_keys_json_pkey'),
response_json TEXT NOT NULL,
server_name TEXT NOT NULL,
valid_until BIGINT NOT NULL
@ -33,7 +33,7 @@ CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_json (
`
const insertServerKeysJSONSQL = "" +
"INSERT INTO federationsender_notary_server_keys_json (response_json, server_name, valid_until) VALUES ($1, $2, $3)" +
"INSERT INTO federationapi_notary_server_keys_json (response_json, server_name, valid_until) VALUES ($1, $2, $3)" +
" RETURNING notary_id"
type notaryServerKeysStatements struct {

32
federationsender/storage/postgres/notary_server_keys_metadata_table.go → federationapi/storage/postgres/notary_server_keys_metadata_table.go
View file

@ -20,13 +20,13 @@ import (
"encoding/json"
"github.com/lib/pq"
"github.com/matrix-org/dendrite/federationsender/storage/tables"
"github.com/matrix-org/dendrite/federationapi/storage/tables"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/gomatrixserverlib"
)
const notaryServerKeysMetadataSchema = `
CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_metadata (
CREATE TABLE IF NOT EXISTS federationapi_notary_server_keys_metadata (
notary_id BIGINT NOT NULL,
server_name TEXT NOT NULL,
key_id TEXT NOT NULL,
@ -35,41 +35,41 @@ CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_metadata (
`
const upsertServerKeysSQL = "" +
"INSERT INTO federationsender_notary_server_keys_metadata (notary_id, server_name, key_id) VALUES ($1, $2, $3)" +
"INSERT INTO federationapi_notary_server_keys_metadata (notary_id, server_name, key_id) VALUES ($1, $2, $3)" +
" ON CONFLICT (server_name, key_id) DO UPDATE SET notary_id = $1"
// for a given (server_name, key_id), find the existing notary ID and valid until. Used to check if we will replace it
// JOINs with the json table
const selectNotaryKeyMetadataSQL = `
SELECT federationsender_notary_server_keys_metadata.notary_id, valid_until FROM federationsender_notary_server_keys_json
JOIN federationsender_notary_server_keys_metadata ON
federationsender_notary_server_keys_metadata.notary_id = federationsender_notary_server_keys_json.notary_id
WHERE federationsender_notary_server_keys_metadata.server_name = $1 AND federationsender_notary_server_keys_metadata.key_id = $2
SELECT federationapi_notary_server_keys_metadata.notary_id, valid_until FROM federationapi_notary_server_keys_json
JOIN federationapi_notary_server_keys_metadata ON
federationapi_notary_server_keys_metadata.notary_id = federationapi_notary_server_keys_json.notary_id
WHERE federationapi_notary_server_keys_metadata.server_name = $1 AND federationapi_notary_server_keys_metadata.key_id = $2
`
// select the response which has the highest valid_until value
// JOINs with the json table
const selectNotaryKeyResponsesSQL = `
SELECT response_json FROM federationsender_notary_server_keys_json
SELECT response_json FROM federationapi_notary_server_keys_json
WHERE server_name = $1 AND valid_until = (
SELECT MAX(valid_until) FROM federationsender_notary_server_keys_json WHERE server_name = $1
SELECT MAX(valid_until) FROM federationapi_notary_server_keys_json WHERE server_name = $1
)
`
// select the responses which have the given key IDs
// JOINs with the json table
const selectNotaryKeyResponsesWithKeyIDsSQL = `
SELECT response_json FROM federationsender_notary_server_keys_json
JOIN federationsender_notary_server_keys_metadata ON
federationsender_notary_server_keys_metadata.notary_id = federationsender_notary_server_keys_json.notary_id
WHERE federationsender_notary_server_keys_json.server_name = $1 AND federationsender_notary_server_keys_metadata.key_id = ANY ($2)
GROUP BY federationsender_notary_server_keys_json.notary_id
SELECT response_json FROM federationapi_notary_server_keys_json
JOIN federationapi_notary_server_keys_metadata ON
federationapi_notary_server_keys_metadata.notary_id = federationapi_notary_server_keys_json.notary_id
WHERE federationapi_notary_server_keys_json.server_name = $1 AND federationapi_notary_server_keys_metadata.key_id = ANY ($2)
GROUP BY federationapi_notary_server_keys_json.notary_id
`
// JOINs with the metadata table
const deleteUnusedServerKeysJSONSQL = `
DELETE FROM federationsender_notary_server_keys_json WHERE federationsender_notary_server_keys_json.notary_id NOT IN (
SELECT DISTINCT notary_id FROM federationsender_notary_server_keys_metadata
DELETE FROM federationapi_notary_server_keys_json WHERE federationapi_notary_server_keys_json.notary_id NOT IN (
SELECT DISTINCT notary_id FROM federationapi_notary_server_keys_metadata
)
`

16
federationsender/storage/postgres/outbound_peeks_table.go → federationapi/storage/postgres/outbound_peeks_table.go
View file

@ -19,14 +19,14 @@ import (
"database/sql"
"time"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
)
const outboundPeeksSchema = `
CREATE TABLE IF NOT EXISTS federationsender_outbound_peeks (
CREATE TABLE IF NOT EXISTS federationapi_outbound_peeks (
room_id TEXT NOT NULL,
server_name TEXT NOT NULL,
peek_id TEXT NOT NULL,
@ -38,22 +38,22 @@ CREATE TABLE IF NOT EXISTS federationsender_outbound_peeks (
`
const insertOutboundPeekSQL = "" +
"INSERT INTO federationsender_outbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
"INSERT INTO federationapi_outbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
const selectOutboundPeekSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_outbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_outbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
const selectOutboundPeeksSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_outbound_peeks WHERE room_id = $1"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_outbound_peeks WHERE room_id = $1"
const renewOutboundPeekSQL = "" +
"UPDATE federationsender_outbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
"UPDATE federationapi_outbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
const deleteOutboundPeekSQL = "" +
"DELETE FROM federationsender_outbound_peeks WHERE room_id = $1 and server_name = $2"
"DELETE FROM federationapi_outbound_peeks WHERE room_id = $1 and server_name = $2"
const deleteOutboundPeeksSQL = "" +
"DELETE FROM federationsender_outbound_peeks WHERE room_id = $1"
"DELETE FROM federationapi_outbound_peeks WHERE room_id = $1"
type outboundPeeksStatements struct {
db *sql.DB

20
federationsender/storage/postgres/queue_edus_table.go → federationapi/storage/postgres/queue_edus_table.go
View file

@ -25,41 +25,41 @@ import (
)
const queueEDUsSchema = `
CREATE TABLE IF NOT EXISTS federationsender_queue_edus (
CREATE TABLE IF NOT EXISTS federationapi_queue_edus (
-- The type of the event (informational).
edu_type TEXT NOT NULL,
-- The domain part of the user ID the EDU event is for.
server_name TEXT NOT NULL,
-- The JSON NID from the federationsender_queue_edus_json table.
-- The JSON NID from the federationapi_queue_edus_json table.
json_nid BIGINT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS federationsender_queue_edus_json_nid_idx
ON federationsender_queue_edus (json_nid, server_name);
CREATE UNIQUE INDEX IF NOT EXISTS federationapi_queue_edus_json_nid_idx
ON federationapi_queue_edus (json_nid, server_name);
`
const insertQueueEDUSQL = "" +
"INSERT INTO federationsender_queue_edus (edu_type, server_name, json_nid)" +
"INSERT INTO federationapi_queue_edus (edu_type, server_name, json_nid)" +
" VALUES ($1, $2, $3)"
const deleteQueueEDUSQL = "" +
"DELETE FROM federationsender_queue_edus WHERE server_name = $1 AND json_nid = ANY($2)"
"DELETE FROM federationapi_queue_edus WHERE server_name = $1 AND json_nid = ANY($2)"
const selectQueueEDUSQL = "" +
"SELECT json_nid FROM federationsender_queue_edus" +
"SELECT json_nid FROM federationapi_queue_edus" +
" WHERE server_name = $1" +
" LIMIT $2"
const selectQueueEDUReferenceJSONCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_edus" +
"SELECT COUNT(*) FROM federationapi_queue_edus" +
" WHERE json_nid = $1"
const selectQueueEDUCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_edus" +
"SELECT COUNT(*) FROM federationapi_queue_edus" +
" WHERE server_name = $1"
const selectQueueServerNamesSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_queue_edus"
"SELECT DISTINCT server_name FROM federationapi_queue_edus"
type queueEDUsStatements struct {
db *sql.DB

12
federationsender/storage/postgres/queue_json_table.go → federationapi/storage/postgres/queue_json_table.go
View file

@ -24,10 +24,10 @@ import (
)
const queueJSONSchema = `
-- The federationsender_queue_json table contains event contents that
-- The federationapi_queue_json table contains event contents that
-- we failed to send.
CREATE TABLE IF NOT EXISTS federationsender_queue_json (
-- The JSON NID. This allows the federationsender_queue_retry table to
CREATE TABLE IF NOT EXISTS federationapi_queue_json (
-- The JSON NID. This allows the federationapi_queue_retry table to
-- cross-reference to find the JSON blob.
json_nid BIGSERIAL,
-- The JSON body. Text so that we preserve UTF-8.
@ -36,15 +36,15 @@ CREATE TABLE IF NOT EXISTS federationsender_queue_json (
`
const insertJSONSQL = "" +
"INSERT INTO federationsender_queue_json (json_body)" +
"INSERT INTO federationapi_queue_json (json_body)" +
" VALUES ($1)" +
" RETURNING json_nid"
const deleteJSONSQL = "" +
"DELETE FROM federationsender_queue_json WHERE json_nid = ANY($1)"
"DELETE FROM federationapi_queue_json WHERE json_nid = ANY($1)"
const selectJSONSQL = "" +
"SELECT json_nid, json_body FROM federationsender_queue_json" +
"SELECT json_nid, json_body FROM federationapi_queue_json" +
" WHERE json_nid = ANY($1)"
type queueJSONStatements struct {

20
federationsender/storage/postgres/queue_pdus_table.go → federationapi/storage/postgres/queue_pdus_table.go
View file

@ -25,41 +25,41 @@ import (
)
const queuePDUsSchema = `
CREATE TABLE IF NOT EXISTS federationsender_queue_pdus (
CREATE TABLE IF NOT EXISTS federationapi_queue_pdus (
-- The transaction ID that was generated before persisting the event.
transaction_id TEXT NOT NULL,
-- The destination server that we will send the event to.
server_name TEXT NOT NULL,
-- The JSON NID from the federationsender_queue_pdus_json table.
-- The JSON NID from the federationapi_queue_pdus_json table.
json_nid BIGINT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS federationsender_queue_pdus_pdus_json_nid_idx
ON federationsender_queue_pdus (json_nid, server_name);
CREATE UNIQUE INDEX IF NOT EXISTS federationapi_queue_pdus_pdus_json_nid_idx
ON federationapi_queue_pdus (json_nid, server_name);
`
const insertQueuePDUSQL = "" +
"INSERT INTO federationsender_queue_pdus (transaction_id, server_name, json_nid)" +
"INSERT INTO federationapi_queue_pdus (transaction_id, server_name, json_nid)" +
" VALUES ($1, $2, $3)"
const deleteQueuePDUSQL = "" +
"DELETE FROM federationsender_queue_pdus WHERE server_name = $1 AND json_nid = ANY($2)"
"DELETE FROM federationapi_queue_pdus WHERE server_name = $1 AND json_nid = ANY($2)"
const selectQueuePDUsSQL = "" +
"SELECT json_nid FROM federationsender_queue_pdus" +
"SELECT json_nid FROM federationapi_queue_pdus" +
" WHERE server_name = $1" +
" LIMIT $2"
const selectQueuePDUReferenceJSONCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_pdus" +
"SELECT COUNT(*) FROM federationapi_queue_pdus" +
" WHERE json_nid = $1"
const selectQueuePDUsCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_pdus" +
"SELECT COUNT(*) FROM federationapi_queue_pdus" +
" WHERE server_name = $1"
const selectQueuePDUServerNamesSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_queue_pdus"
"SELECT DISTINCT server_name FROM federationapi_queue_pdus"
type queuePDUsStatements struct {
db *sql.DB

6
federationsender/storage/postgres/storage.go → federationapi/storage/postgres/storage.go
View file

@ -19,8 +19,8 @@ import (
"database/sql"
"fmt"
"github.com/matrix-org/dendrite/federationsender/storage/postgres/deltas"
"github.com/matrix-org/dendrite/federationsender/storage/shared"
"github.com/matrix-org/dendrite/federationapi/storage/postgres/deltas"
"github.com/matrix-org/dendrite/federationapi/storage/shared"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/config"
@ -97,7 +97,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
NotaryServerKeysJSON: notaryJSON,
NotaryServerKeysMetadata: notaryMetadata,
}
if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "federationsender"); err != nil {
if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "federationapi"); err != nil {
return nil, err
}
return &d, nil

4
federationsender/storage/shared/storage.go → federationapi/storage/shared/storage.go
View file

@ -20,8 +20,8 @@ import (
"fmt"
"time"
"github.com/matrix-org/dendrite/federationsender/storage/tables"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/storage/tables"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"

2
federationsender/storage/shared/storage_edus.go → federationapi/storage/shared/storage_edus.go
View file

@ -38,7 +38,7 @@ func (d *Database) AssociateEDUWithDestination(
txn, // SQL transaction
"", // TODO: EDU type for coalescing
serverName, // destination server name
receipt.nid, // NID from the federationsender_queue_json table
receipt.nid, // NID from the federationapi_queue_json table
); err != nil {
return fmt.Errorf("InsertQueueEDU: %w", err)
}

2
federationsender/storage/shared/storage_pdus.go → federationapi/storage/shared/storage_pdus.go
View file

@ -39,7 +39,7 @@ func (d *Database) AssociatePDUWithDestination(
txn, // SQL transaction
transactionID, // transaction ID
serverName, // destination server name
receipt.nid, // NID from the federationsender_queue_json table
receipt.nid, // NID from the federationapi_queue_json table
); err != nil {
return fmt.Errorf("InsertQueuePDU: %w", err)
}

10
federationsender/storage/sqlite3/blacklist_table.go → federationapi/storage/sqlite3/blacklist_table.go
View file

@ -23,7 +23,7 @@ import (
)
const blacklistSchema = `
CREATE TABLE IF NOT EXISTS federationsender_blacklist (
CREATE TABLE IF NOT EXISTS federationapi_blacklist (
-- The blacklisted server name
server_name TEXT NOT NULL,
UNIQUE (server_name)
@ -31,17 +31,17 @@ CREATE TABLE IF NOT EXISTS federationsender_blacklist (
`
const insertBlacklistSQL = "" +
"INSERT INTO federationsender_blacklist (server_name) VALUES ($1)" +
"INSERT INTO federationapi_blacklist (server_name) VALUES ($1)" +
" ON CONFLICT DO NOTHING"
const selectBlacklistSQL = "" +
"SELECT server_name FROM federationsender_blacklist WHERE server_name = $1"
"SELECT server_name FROM federationapi_blacklist WHERE server_name = $1"
const deleteBlacklistSQL = "" +
"DELETE FROM federationsender_blacklist WHERE server_name = $1"
"DELETE FROM federationapi_blacklist WHERE server_name = $1"
const deleteAllBlacklistSQL = "" +
"DELETE FROM federationsender_blacklist"
"DELETE FROM federationapi_blacklist"
type blacklistStatements struct {
db *sql.DB

2
federationsender/storage/sqlite3/deltas/2021020411080000_rooms.go → federationapi/storage/sqlite3/deltas/2021020411080000_rooms.go
View file

@ -32,7 +32,7 @@ func LoadRemoveRoomsTable(m *sqlutil.Migrations) {
func UpRemoveRoomsTable(tx *sql.Tx) error {
_, err := tx.Exec(`
DROP TABLE IF EXISTS federationsender_rooms;
DROP TABLE IF EXISTS federationapi_rooms;
`)
if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err)

16
federationsender/storage/sqlite3/inbound_peeks_table.go → federationapi/storage/sqlite3/inbound_peeks_table.go
View file

@ -19,14 +19,14 @@ import (
"database/sql"
"time"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
)
const inboundPeeksSchema = `
CREATE TABLE IF NOT EXISTS federationsender_inbound_peeks (
CREATE TABLE IF NOT EXISTS federationapi_inbound_peeks (
room_id TEXT NOT NULL,
server_name TEXT NOT NULL,
peek_id TEXT NOT NULL,
@ -38,22 +38,22 @@ CREATE TABLE IF NOT EXISTS federationsender_inbound_peeks (
`
const insertInboundPeekSQL = "" +
"INSERT INTO federationsender_inbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
"INSERT INTO federationapi_inbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
const selectInboundPeekSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_inbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_inbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
const selectInboundPeeksSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_inbound_peeks WHERE room_id = $1"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_inbound_peeks WHERE room_id = $1"
const renewInboundPeekSQL = "" +
"UPDATE federationsender_inbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
"UPDATE federationapi_inbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
const deleteInboundPeekSQL = "" +
"DELETE FROM federationsender_inbound_peeks WHERE room_id = $1 and server_name = $2"
"DELETE FROM federationapi_inbound_peeks WHERE room_id = $1 and server_name = $2"
const deleteInboundPeeksSQL = "" +
"DELETE FROM federationsender_inbound_peeks WHERE room_id = $1"
"DELETE FROM federationapi_inbound_peeks WHERE room_id = $1"
type inboundPeeksStatements struct {
db *sql.DB

20
federationsender/storage/sqlite3/joined_hosts_table.go → federationapi/storage/sqlite3/joined_hosts_table.go
View file

@ -20,7 +20,7 @@ import (
"database/sql"
"strings"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
@ -30,7 +30,7 @@ const joinedHostsSchema = `
-- The joined_hosts table stores a list of m.room.member event ids in the
-- current state for each room where the membership is "join".
-- There will be an entry for every user that is joined to the room.
CREATE TABLE IF NOT EXISTS federationsender_joined_hosts (
CREATE TABLE IF NOT EXISTS federationapi_joined_hosts (
-- The string ID of the room.
room_id TEXT NOT NULL,
-- The event ID of the m.room.member join event.
@ -40,31 +40,31 @@ CREATE TABLE IF NOT EXISTS federationsender_joined_hosts (
);
CREATE UNIQUE INDEX IF NOT EXISTS federatonsender_joined_hosts_event_id_idx
ON federationsender_joined_hosts (event_id);
ON federationapi_joined_hosts (event_id);
CREATE INDEX IF NOT EXISTS federatonsender_joined_hosts_room_id_idx
ON federationsender_joined_hosts (room_id)
ON federationapi_joined_hosts (room_id)
`
const insertJoinedHostsSQL = "" +
"INSERT OR IGNORE INTO federationsender_joined_hosts (room_id, event_id, server_name)" +
"INSERT OR IGNORE INTO federationapi_joined_hosts (room_id, event_id, server_name)" +
" VALUES ($1, $2, $3)"
const deleteJoinedHostsSQL = "" +
"DELETE FROM federationsender_joined_hosts WHERE event_id = $1"
"DELETE FROM federationapi_joined_hosts WHERE event_id = $1"
const deleteJoinedHostsForRoomSQL = "" +
"DELETE FROM federationsender_joined_hosts WHERE room_id = $1"
"DELETE FROM federationapi_joined_hosts WHERE room_id = $1"
const selectJoinedHostsSQL = "" +
"SELECT event_id, server_name FROM federationsender_joined_hosts" +
"SELECT event_id, server_name FROM federationapi_joined_hosts" +
" WHERE room_id = $1"
const selectAllJoinedHostsSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_joined_hosts"
"SELECT DISTINCT server_name FROM federationapi_joined_hosts"
const selectJoinedHostsForRoomsSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_joined_hosts WHERE room_id IN ($1)"
"SELECT DISTINCT server_name FROM federationapi_joined_hosts WHERE room_id IN ($1)"
type joinedHostsStatements struct {
db *sql.DB

6
federationsender/storage/sqlite3/notary_server_keys_json_table.go → federationapi/storage/sqlite3/notary_server_keys_json_table.go
View file

@ -18,12 +18,12 @@ import (
"context"
"database/sql"
"github.com/matrix-org/dendrite/federationsender/storage/tables"
"github.com/matrix-org/dendrite/federationapi/storage/tables"
"github.com/matrix-org/gomatrixserverlib"
)
const notaryServerKeysJSONSchema = `
CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_json (
CREATE TABLE IF NOT EXISTS federationapi_notary_server_keys_json (
notary_id INTEGER PRIMARY KEY AUTOINCREMENT,
response_json TEXT NOT NULL,
server_name TEXT NOT NULL,
@ -32,7 +32,7 @@ CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_json (
`
const insertServerKeysJSONSQL = "" +
"INSERT INTO federationsender_notary_server_keys_json (response_json, server_name, valid_until) VALUES ($1, $2, $3)" +
"INSERT INTO federationapi_notary_server_keys_json (response_json, server_name, valid_until) VALUES ($1, $2, $3)" +
" RETURNING notary_id"
type notaryServerKeysStatements struct {

32
federationsender/storage/sqlite3/notary_server_keys_metadata_table.go → federationapi/storage/sqlite3/notary_server_keys_metadata_table.go
View file

@ -21,14 +21,14 @@ import (
"fmt"
"strings"
"github.com/matrix-org/dendrite/federationsender/storage/tables"
"github.com/matrix-org/dendrite/federationapi/storage/tables"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
)
const notaryServerKeysMetadataSchema = `
CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_metadata (
CREATE TABLE IF NOT EXISTS federationapi_notary_server_keys_metadata (
notary_id BIGINT NOT NULL,
server_name TEXT NOT NULL,
key_id TEXT NOT NULL,
@ -37,41 +37,41 @@ CREATE TABLE IF NOT EXISTS federationsender_notary_server_keys_metadata (
`
const upsertServerKeysSQL = "" +
"INSERT INTO federationsender_notary_server_keys_metadata (notary_id, server_name, key_id) VALUES ($1, $2, $3)" +
"INSERT INTO federationapi_notary_server_keys_metadata (notary_id, server_name, key_id) VALUES ($1, $2, $3)" +
" ON CONFLICT (server_name, key_id) DO UPDATE SET notary_id = $1"
// for a given (server_name, key_id), find the existing notary ID and valid until. Used to check if we will replace it
// JOINs with the json table
const selectNotaryKeyMetadataSQL = `
SELECT federationsender_notary_server_keys_metadata.notary_id, valid_until FROM federationsender_notary_server_keys_json
JOIN federationsender_notary_server_keys_metadata ON
federationsender_notary_server_keys_metadata.notary_id = federationsender_notary_server_keys_json.notary_id
WHERE federationsender_notary_server_keys_metadata.server_name = $1 AND federationsender_notary_server_keys_metadata.key_id = $2
SELECT federationapi_notary_server_keys_metadata.notary_id, valid_until FROM federationapi_notary_server_keys_json
JOIN federationapi_notary_server_keys_metadata ON
federationapi_notary_server_keys_metadata.notary_id = federationapi_notary_server_keys_json.notary_id
WHERE federationapi_notary_server_keys_metadata.server_name = $1 AND federationapi_notary_server_keys_metadata.key_id = $2
`
// select the response which has the highest valid_until value
// JOINs with the json table
const selectNotaryKeyResponsesSQL = `
SELECT response_json FROM federationsender_notary_server_keys_json
SELECT response_json FROM federationapi_notary_server_keys_json
WHERE server_name = $1 AND valid_until = (
SELECT MAX(valid_until) FROM federationsender_notary_server_keys_json WHERE server_name = $1
SELECT MAX(valid_until) FROM federationapi_notary_server_keys_json WHERE server_name = $1
)
`
// select the responses which have the given key IDs
// JOINs with the json table
const selectNotaryKeyResponsesWithKeyIDsSQL = `
SELECT response_json FROM federationsender_notary_server_keys_json
JOIN federationsender_notary_server_keys_metadata ON
federationsender_notary_server_keys_metadata.notary_id = federationsender_notary_server_keys_json.notary_id
WHERE federationsender_notary_server_keys_json.server_name = $1 AND federationsender_notary_server_keys_metadata.key_id IN ($2)
GROUP BY federationsender_notary_server_keys_json.notary_id
SELECT response_json FROM federationapi_notary_server_keys_json
JOIN federationapi_notary_server_keys_metadata ON
federationapi_notary_server_keys_metadata.notary_id = federationapi_notary_server_keys_json.notary_id
WHERE federationapi_notary_server_keys_json.server_name = $1 AND federationapi_notary_server_keys_metadata.key_id IN ($2)
GROUP BY federationapi_notary_server_keys_json.notary_id
`
// JOINs with the metadata table
const deleteUnusedServerKeysJSONSQL = `
DELETE FROM federationsender_notary_server_keys_json WHERE federationsender_notary_server_keys_json.notary_id NOT IN (
SELECT DISTINCT notary_id FROM federationsender_notary_server_keys_metadata
DELETE FROM federationapi_notary_server_keys_json WHERE federationapi_notary_server_keys_json.notary_id NOT IN (
SELECT DISTINCT notary_id FROM federationapi_notary_server_keys_metadata
)
`

16
federationsender/storage/sqlite3/outbound_peeks_table.go → federationapi/storage/sqlite3/outbound_peeks_table.go
View file

@ -19,14 +19,14 @@ import (
"database/sql"
"time"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
)
const outboundPeeksSchema = `
CREATE TABLE IF NOT EXISTS federationsender_outbound_peeks (
CREATE TABLE IF NOT EXISTS federationapi_outbound_peeks (
room_id TEXT NOT NULL,
server_name TEXT NOT NULL,
peek_id TEXT NOT NULL,
@ -38,22 +38,22 @@ CREATE TABLE IF NOT EXISTS federationsender_outbound_peeks (
`
const insertOutboundPeekSQL = "" +
"INSERT INTO federationsender_outbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
"INSERT INTO federationapi_outbound_peeks (room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval) VALUES ($1, $2, $3, $4, $5, $6)"
const selectOutboundPeekSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_outbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_outbound_peeks WHERE room_id = $1 and server_name = $2 and peek_id = $3"
const selectOutboundPeeksSQL = "" +
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationsender_outbound_peeks WHERE room_id = $1"
"SELECT room_id, server_name, peek_id, creation_ts, renewed_ts, renewal_interval FROM federationapi_outbound_peeks WHERE room_id = $1"
const renewOutboundPeekSQL = "" +
"UPDATE federationsender_outbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
"UPDATE federationapi_outbound_peeks SET renewed_ts=$1, renewal_interval=$2 WHERE room_id = $3 and server_name = $4 and peek_id = $5"
const deleteOutboundPeekSQL = "" +
"DELETE FROM federationsender_outbound_peeks WHERE room_id = $1 and server_name = $2"
"DELETE FROM federationapi_outbound_peeks WHERE room_id = $1 and server_name = $2"
const deleteOutboundPeeksSQL = "" +
"DELETE FROM federationsender_outbound_peeks WHERE room_id = $1"
"DELETE FROM federationapi_outbound_peeks WHERE room_id = $1"
type outboundPeeksStatements struct {
db *sql.DB

20
federationsender/storage/sqlite3/queue_edus_table.go → federationapi/storage/sqlite3/queue_edus_table.go
View file

@ -26,41 +26,41 @@ import (
)
const queueEDUsSchema = `
CREATE TABLE IF NOT EXISTS federationsender_queue_edus (
CREATE TABLE IF NOT EXISTS federationapi_queue_edus (
-- The type of the event (informational).
edu_type TEXT NOT NULL,
-- The domain part of the user ID the EDU event is for.
server_name TEXT NOT NULL,
-- The JSON NID from the federationsender_queue_edus_json table.
-- The JSON NID from the federationapi_queue_edus_json table.
json_nid BIGINT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS federationsender_queue_edus_json_nid_idx
ON federationsender_queue_edus (json_nid, server_name);
CREATE UNIQUE INDEX IF NOT EXISTS federationapi_queue_edus_json_nid_idx
ON federationapi_queue_edus (json_nid, server_name);
`
const insertQueueEDUSQL = "" +
"INSERT INTO federationsender_queue_edus (edu_type, server_name, json_nid)" +
"INSERT INTO federationapi_queue_edus (edu_type, server_name, json_nid)" +
" VALUES ($1, $2, $3)"
const deleteQueueEDUsSQL = "" +
"DELETE FROM federationsender_queue_edus WHERE server_name = $1 AND json_nid IN ($2)"
"DELETE FROM federationapi_queue_edus WHERE server_name = $1 AND json_nid IN ($2)"
const selectQueueEDUSQL = "" +
"SELECT json_nid FROM federationsender_queue_edus" +
"SELECT json_nid FROM federationapi_queue_edus" +
" WHERE server_name = $1" +
" LIMIT $2"
const selectQueueEDUReferenceJSONCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_edus" +
"SELECT COUNT(*) FROM federationapi_queue_edus" +
" WHERE json_nid = $1"
const selectQueueEDUCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_edus" +
"SELECT COUNT(*) FROM federationapi_queue_edus" +
" WHERE server_name = $1"
const selectQueueServerNamesSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_queue_edus"
"SELECT DISTINCT server_name FROM federationapi_queue_edus"
type queueEDUsStatements struct {
db *sql.DB

10
federationsender/storage/sqlite3/queue_json_table.go → federationapi/storage/sqlite3/queue_json_table.go
View file

@ -28,8 +28,8 @@ import (
const queueJSONSchema = `
-- The queue_retry_json table contains event contents that
-- we failed to send.
CREATE TABLE IF NOT EXISTS federationsender_queue_json (
-- The JSON NID. This allows the federationsender_queue_retry table to
CREATE TABLE IF NOT EXISTS federationapi_queue_json (
-- The JSON NID. This allows the federationapi_queue_retry table to
-- cross-reference to find the JSON blob.
json_nid INTEGER PRIMARY KEY AUTOINCREMENT,
-- The JSON body. Text so that we preserve UTF-8.
@ -38,14 +38,14 @@ CREATE TABLE IF NOT EXISTS federationsender_queue_json (
`
const insertJSONSQL = "" +
"INSERT INTO federationsender_queue_json (json_body)" +
"INSERT INTO federationapi_queue_json (json_body)" +
" VALUES ($1)"
const deleteJSONSQL = "" +
"DELETE FROM federationsender_queue_json WHERE json_nid IN ($1)"
"DELETE FROM federationapi_queue_json WHERE json_nid IN ($1)"
const selectJSONSQL = "" +
"SELECT json_nid, json_body FROM federationsender_queue_json" +
"SELECT json_nid, json_body FROM federationapi_queue_json" +
" WHERE json_nid IN ($1)"
type queueJSONStatements struct {

22
federationsender/storage/sqlite3/queue_pdus_table.go → federationapi/storage/sqlite3/queue_pdus_table.go
View file

@ -27,47 +27,47 @@ import (
)
const queuePDUsSchema = `
CREATE TABLE IF NOT EXISTS federationsender_queue_pdus (
CREATE TABLE IF NOT EXISTS federationapi_queue_pdus (
-- The transaction ID that was generated before persisting the event.
transaction_id TEXT NOT NULL,
-- The domain part of the user ID the m.room.member event is for.
server_name TEXT NOT NULL,
-- The JSON NID from the federationsender_queue_pdus_json table.
-- The JSON NID from the federationapi_queue_pdus_json table.
json_nid BIGINT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS federationsender_queue_pdus_pdus_json_nid_idx
ON federationsender_queue_pdus (json_nid, server_name);
CREATE UNIQUE INDEX IF NOT EXISTS federationapi_queue_pdus_pdus_json_nid_idx
ON federationapi_queue_pdus (json_nid, server_name);
`
const insertQueuePDUSQL = "" +
"INSERT INTO federationsender_queue_pdus (transaction_id, server_name, json_nid)" +
"INSERT INTO federationapi_queue_pdus (transaction_id, server_name, json_nid)" +
" VALUES ($1, $2, $3)"
const deleteQueuePDUsSQL = "" +
"DELETE FROM federationsender_queue_pdus WHERE server_name = $1 AND json_nid IN ($2)"
"DELETE FROM federationapi_queue_pdus WHERE server_name = $1 AND json_nid IN ($2)"
const selectQueueNextTransactionIDSQL = "" +
"SELECT transaction_id FROM federationsender_queue_pdus" +
"SELECT transaction_id FROM federationapi_queue_pdus" +
" WHERE server_name = $1" +
" ORDER BY transaction_id ASC" +
" LIMIT 1"
const selectQueuePDUsSQL = "" +
"SELECT json_nid FROM federationsender_queue_pdus" +
"SELECT json_nid FROM federationapi_queue_pdus" +
" WHERE server_name = $1" +
" LIMIT $2"
const selectQueuePDUsReferenceJSONCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_pdus" +
"SELECT COUNT(*) FROM federationapi_queue_pdus" +
" WHERE json_nid = $1"
const selectQueuePDUsCountSQL = "" +
"SELECT COUNT(*) FROM federationsender_queue_pdus" +
"SELECT COUNT(*) FROM federationapi_queue_pdus" +
" WHERE server_name = $1"
const selectQueuePDUsServerNamesSQL = "" +
"SELECT DISTINCT server_name FROM federationsender_queue_pdus"
"SELECT DISTINCT server_name FROM federationapi_queue_pdus"
type queuePDUsStatements struct {
db *sql.DB

6
federationsender/storage/sqlite3/storage.go → federationapi/storage/sqlite3/storage.go
View file

@ -18,8 +18,8 @@ package sqlite3
import (
"database/sql"
"github.com/matrix-org/dendrite/federationsender/storage/shared"
"github.com/matrix-org/dendrite/federationsender/storage/sqlite3/deltas"
"github.com/matrix-org/dendrite/federationapi/storage/shared"
"github.com/matrix-org/dendrite/federationapi/storage/sqlite3/deltas"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/config"
@ -96,7 +96,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
NotaryServerKeysJSON: notaryKeys,
NotaryServerKeysMetadata: notaryKeysMetadata,
}
if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "federationsender"); err != nil {
if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "federationapi"); err != nil {
return nil, err
}
return &d, nil

4
federationsender/storage/storage.go → federationapi/storage/storage.go
View file

@ -20,8 +20,8 @@ package storage
import (
"fmt"
"github.com/matrix-org/dendrite/federationsender/storage/postgres"
"github.com/matrix-org/dendrite/federationsender/storage/sqlite3"
"github.com/matrix-org/dendrite/federationapi/storage/postgres"
"github.com/matrix-org/dendrite/federationapi/storage/sqlite3"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/setup/config"
)

2
federationsender/storage/storage_wasm.go → federationapi/storage/storage_wasm.go
View file

@ -17,7 +17,7 @@ package storage
import (
"fmt"
"github.com/matrix-org/dendrite/federationsender/storage/sqlite3"
"github.com/matrix-org/dendrite/federationapi/storage/sqlite3"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/setup/config"
)

2
federationsender/storage/tables/interface.go → federationapi/storage/tables/interface.go
View file

@ -18,7 +18,7 @@ import (
"context"
"database/sql"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/gomatrixserverlib"
)

0
federationsender/types/types.go → federationapi/types/types.go
View file

99
federationsender/federationsender.go
View file

@ -1,99 +0,0 @@
// Copyright 2017 Vector Creations Ltd
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package federationsender
import (
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/federationsender/api"
"github.com/matrix-org/dendrite/federationsender/consumers"
"github.com/matrix-org/dendrite/federationsender/internal"
"github.com/matrix-org/dendrite/federationsender/inthttp"
"github.com/matrix-org/dendrite/federationsender/queue"
"github.com/matrix-org/dendrite/federationsender/statistics"
"github.com/matrix-org/dendrite/federationsender/storage"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/kafka"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
)
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
// on the given input API.
func AddInternalRoutes(router *mux.Router, intAPI api.FederationSenderInternalAPI) {
inthttp.AddRoutes(intAPI, router)
}
// NewInternalAPI returns a concerete implementation of the internal API. Callers
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
func NewInternalAPI(
base *setup.BaseDendrite,
federation *gomatrixserverlib.FederationClient,
rsAPI roomserverAPI.RoomserverInternalAPI,
keyRing *gomatrixserverlib.KeyRing,
resetBlacklist bool,
) api.FederationSenderInternalAPI {
cfg := &base.Cfg.FederationSender
federationSenderDB, err := storage.NewDatabase(&cfg.Database, base.Caches)
if err != nil {
logrus.WithError(err).Panic("failed to connect to federation sender db")
}
if resetBlacklist {
_ = federationSenderDB.RemoveAllServersFromBlacklist()
}
stats := &statistics.Statistics{
DB: federationSenderDB,
FailuresUntilBlacklist: cfg.FederationMaxRetries,
}
consumer, _ := kafka.SetupConsumerProducer(&cfg.Matrix.Kafka)
queues := queue.NewOutgoingQueues(
federationSenderDB, base.ProcessContext,
cfg.Matrix.DisableFederation,
cfg.Matrix.ServerName, federation, rsAPI, stats,
&queue.SigningInfo{
KeyID: cfg.Matrix.KeyID,
PrivateKey: cfg.Matrix.PrivateKey,
ServerName: cfg.Matrix.ServerName,
},
)
rsConsumer := consumers.NewOutputRoomEventConsumer(
base.ProcessContext, cfg, consumer, queues,
federationSenderDB, rsAPI,
)
if err = rsConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start room server consumer")
}
tsConsumer := consumers.NewOutputEDUConsumer(
base.ProcessContext, cfg, consumer, queues, federationSenderDB,
)
if err := tsConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start typing server consumer")
}
keyConsumer := consumers.NewKeyChangeConsumer(
base.ProcessContext, &base.Cfg.KeyServer, consumer, queues, federationSenderDB, rsAPI,
)
if err := keyConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start key server consumer")
}
return internal.NewFederationSenderInternalAPI(federationSenderDB, cfg, rsAPI, federation, keyRing, stats, queues)
}

8
internal/httputil/httpapi.go
View file

@ -29,7 +29,7 @@ import (
"github.com/getsentry/sentry-go"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/clientapi/auth"
federationsenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationapiAPI "github.com/matrix-org/dendrite/federationapi/api"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@ -251,7 +251,7 @@ func MakeFedAPI(
}
type FederationWakeups struct {
FsAPI federationsenderAPI.FederationSenderInternalAPI
FsAPI federationapiAPI.FederationInternalAPI
origins sync.Map
}
@ -263,10 +263,10 @@ func (f *FederationWakeups) Wakeup(ctx context.Context, origin gomatrixserverlib
return
}
}
aliveReq := federationsenderAPI.PerformServersAliveRequest{
aliveReq := federationapiAPI.PerformServersAliveRequest{
Servers: []gomatrixserverlib.ServerName{origin},
}
aliveRes := federationsenderAPI.PerformServersAliveResponse{}
aliveRes := federationapiAPI.PerformServersAliveResponse{}
if err := f.FsAPI.PerformServersAlive(ctx, &aliveReq, &aliveRes); err != nil {
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
"origin": origin,

2
keyserver/internal/device_list_update.go
View file

@ -22,7 +22,7 @@ import (
"sync"
"time"
fedsenderapi "github.com/matrix-org/dendrite/federationsender/api"
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"

2
keyserver/internal/internal.go
View file

@ -22,7 +22,7 @@ import (
"sync"
"time"
fedsenderapi "github.com/matrix-org/dendrite/federationsender/api"
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/keyserver/producers"
"github.com/matrix-org/dendrite/keyserver/storage"

2
keyserver/keyserver.go
View file

@ -16,7 +16,7 @@ package keyserver
import (
"github.com/gorilla/mux"
fedsenderapi "github.com/matrix-org/dendrite/federationsender/api"
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/keyserver/consumers"
"github.com/matrix-org/dendrite/keyserver/internal"

2
roomserver/api/api.go
View file

@ -4,7 +4,7 @@ import (
"context"
asAPI "github.com/matrix-org/dendrite/appservice/api"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
)
// RoomserverInputAPI is used to write events to the room server.

2
roomserver/api/api_trace.go
View file

@ -6,7 +6,7 @@ import (
"fmt"
asAPI "github.com/matrix-org/dendrite/appservice/api"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/util"
)

6
roomserver/internal/api.go
View file

@ -6,7 +6,7 @@ import (
"github.com/Shopify/sarama"
"github.com/getsentry/sentry-go"
asAPI "github.com/matrix-org/dendrite/appservice/api"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/roomserver/acls"
"github.com/matrix-org/dendrite/roomserver/api"
@ -37,7 +37,7 @@ type RoomserverInternalAPI struct {
Cache caching.RoomServerCaches
ServerName gomatrixserverlib.ServerName
KeyRing gomatrixserverlib.JSONVerifier
fsAPI fsAPI.FederationSenderInternalAPI
fsAPI fsAPI.FederationInternalAPI
asAPI asAPI.AppServiceQueryAPI
OutputRoomEventTopic string // Kafka topic for new output room events
PerspectiveServerNames []gomatrixserverlib.ServerName
@ -77,7 +77,7 @@ func NewRoomserverAPI(
// SetFederationSenderInputAPI passes in a federation sender input API reference
// so that we can avoid the chicken-and-egg problem of both the roomserver input API
// and the federation sender input API being interdependent.
func (r *RoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationSenderInternalAPI) {
func (r *RoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationInternalAPI) {
r.fsAPI = fsAPI
r.Inviter = &perform.Inviter{

8
roomserver/internal/perform/perform_backfill.go
View file

@ -18,7 +18,7 @@ import (
"context"
"fmt"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/auth"
@ -38,7 +38,7 @@ const maxBackfillServers = 5
type Backfiller struct {
ServerName gomatrixserverlib.ServerName
DB storage.Database
FSAPI federationSenderAPI.FederationSenderInternalAPI
FSAPI federationSenderAPI.FederationInternalAPI
KeyRing gomatrixserverlib.JSONVerifier
// The servers which should be preferred above other servers when backfilling
@ -224,7 +224,7 @@ func (r *Backfiller) fetchAndStoreMissingEvents(ctx context.Context, roomVer gom
// backfillRequester implements gomatrixserverlib.BackfillRequester
type backfillRequester struct {
db storage.Database
fsAPI federationSenderAPI.FederationSenderInternalAPI
fsAPI federationSenderAPI.FederationInternalAPI
thisServer gomatrixserverlib.ServerName
preferServer map[gomatrixserverlib.ServerName]bool
bwExtrems map[string][]string
@ -236,7 +236,7 @@ type backfillRequester struct {
}
func newBackfillRequester(
db storage.Database, fsAPI federationSenderAPI.FederationSenderInternalAPI, thisServer gomatrixserverlib.ServerName,
db storage.Database, fsAPI federationSenderAPI.FederationInternalAPI, thisServer gomatrixserverlib.ServerName,
bwExtrems map[string][]string, preferServers []gomatrixserverlib.ServerName,
) *backfillRequester {
preferServer := make(map[gomatrixserverlib.ServerName]bool)

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

@ -34,7 +34,7 @@ type InboundPeeker struct {
}
// PerformInboundPeek handles peeking into matrix rooms, including over
// federation by talking to the federationsender. called when a remote server
// federation by talking to the federationapi. called when a remote server
// initiates a /peek over federation.
//
// It should atomically figure out the current state of the room (for the

4
roomserver/internal/perform/perform_invite.go
View file

@ -18,7 +18,7 @@ import (
"context"
"fmt"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/internal/input"
@ -33,7 +33,7 @@ import (
type Inviter struct {
DB storage.Database
Cfg *config.RoomServer
FSAPI federationSenderAPI.FederationSenderInternalAPI
FSAPI federationSenderAPI.FederationInternalAPI
Inputer *input.Inputer
}

6
roomserver/internal/perform/perform_join.go
View file

@ -22,7 +22,7 @@ import (
"time"
"github.com/getsentry/sentry-go"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/eventutil"
rsAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
@ -37,7 +37,7 @@ import (
type Joiner struct {
ServerName gomatrixserverlib.ServerName
Cfg *config.RoomServer
FSAPI fsAPI.FederationSenderInternalAPI
FSAPI fsAPI.FederationInternalAPI
RSAPI rsAPI.RoomserverInternalAPI
DB storage.Database
@ -45,7 +45,7 @@ type Joiner struct {
Queryer *query.Queryer
}
// PerformJoin handles joining matrix rooms, including over federation by talking to the federationsender.
// PerformJoin handles joining matrix rooms, including over federation by talking to the federationapi.
func (r *Joiner) PerformJoin(
ctx context.Context,
req *rsAPI.PerformJoinRequest,

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

@ -19,7 +19,7 @@ import (
"fmt"
"strings"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/helpers"
"github.com/matrix-org/dendrite/roomserver/internal/input"
@ -32,7 +32,7 @@ import (
type Leaver struct {
Cfg *config.RoomServer
DB storage.Database
FSAPI fsAPI.FederationSenderInternalAPI
FSAPI fsAPI.FederationInternalAPI
Inputer *input.Inputer
}

6
roomserver/internal/perform/perform_peek.go
View file

@ -20,7 +20,7 @@ import (
"fmt"
"strings"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/dendrite/roomserver/storage"
@ -33,13 +33,13 @@ import (
type Peeker struct {
ServerName gomatrixserverlib.ServerName
Cfg *config.RoomServer
FSAPI fsAPI.FederationSenderInternalAPI
FSAPI fsAPI.FederationInternalAPI
DB storage.Database
Inputer *input.Inputer
}
// PerformPeek handles peeking into matrix rooms, including over federation by talking to the federationsender.
// PerformPeek handles peeking into matrix rooms, including over federation by talking to the federationapi.
func (r *Peeker) PerformPeek(
ctx context.Context,
req *api.PerformPeekRequest,

6
roomserver/internal/perform/perform_unpeek.go
View file

@ -19,7 +19,7 @@ import (
"fmt"
"strings"
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
fsAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal/input"
"github.com/matrix-org/dendrite/roomserver/storage"
@ -30,13 +30,13 @@ import (
type Unpeeker struct {
ServerName gomatrixserverlib.ServerName
Cfg *config.RoomServer
FSAPI fsAPI.FederationSenderInternalAPI
FSAPI fsAPI.FederationInternalAPI
DB storage.Database
Inputer *input.Inputer
}
// PerformPeek handles peeking into matrix rooms, including over federation by talking to the federationsender.
// PerformPeek handles peeking into matrix rooms, including over federation by talking to the federationapi.
func (r *Unpeeker) PerformUnpeek(
ctx context.Context,
req *api.PerformUnpeekRequest,

4
roomserver/inthttp/client.go
View file

@ -7,7 +7,7 @@ import (
"net/http"
asAPI "github.com/matrix-org/dendrite/appservice/api"
fsInputAPI "github.com/matrix-org/dendrite/federationsender/api"
fsInputAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/roomserver/api"
@ -83,7 +83,7 @@ func NewRoomserverClient(
}
// SetFederationSenderInputAPI no-ops in HTTP client mode as there is no chicken/egg scenario
func (h *httpRoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsInputAPI.FederationSenderInternalAPI) {
func (h *httpRoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsInputAPI.FederationInternalAPI) {
}
// SetAppserviceAPI no-ops in HTTP client mode as there is no chicken/egg scenario

8
setup/base.go
View file

@ -47,8 +47,8 @@ import (
asinthttp "github.com/matrix-org/dendrite/appservice/inthttp"
eduServerAPI "github.com/matrix-org/dendrite/eduserver/api"
eduinthttp "github.com/matrix-org/dendrite/eduserver/inthttp"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
fsinthttp "github.com/matrix-org/dendrite/federationsender/inthttp"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
fsinthttp "github.com/matrix-org/dendrite/federationapi/inthttp"
keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
keyinthttp "github.com/matrix-org/dendrite/keyserver/inthttp"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
@ -248,9 +248,9 @@ func (b *BaseDendrite) EDUServerClient() eduServerAPI.EDUServerInputAPI {
return e
}
// FederationSenderHTTPClient returns FederationSenderInternalAPI for hitting
// FederationSenderHTTPClient returns FederationInternalAPI for hitting
// the federation sender over HTTP
func (b *BaseDendrite) FederationSenderHTTPClient() federationSenderAPI.FederationSenderInternalAPI {
func (b *BaseDendrite) FederationSenderHTTPClient() federationSenderAPI.FederationInternalAPI {
f, err := fsinthttp.NewFederationSenderClient(b.Cfg.FederationSenderURL(), b.apiHttpClient)
if err != nil {
logrus.WithError(err).Panic("FederationSenderHTTPClient failed", b.apiHttpClient)

2
setup/config/config_federationsender.go
View file

@ -26,7 +26,7 @@ func (c *FederationSender) Defaults() {
c.InternalAPI.Listen = "http://localhost:7775"
c.InternalAPI.Connect = "http://localhost:7775"
c.Database.Defaults(10)
c.Database.ConnectionString = "file:federationsender.db"
c.Database.ConnectionString = "file:federationapi.db"
c.FederationMaxRetries = 16
c.DisableTLSValidation = false

2
setup/config/config_test.go
View file

@ -112,7 +112,7 @@ federation_sender:
listen: http://localhost:7775
connect: http://localhost:7775
database:
connection_string: file:federationsender.db
connection_string: file:federationapi.db
max_open_conns: 100
max_idle_conns: 2
conn_max_lifetime: -1

4
setup/monolith.go
View file

@ -21,7 +21,7 @@ import (
"github.com/matrix-org/dendrite/clientapi/api"
eduServerAPI "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/federationapi"
federationSenderAPI "github.com/matrix-org/dendrite/federationsender/api"
federationSenderAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/transactions"
keyAPI "github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/mediaapi"
@ -46,7 +46,7 @@ type Monolith struct {
AppserviceAPI appserviceAPI.AppServiceQueryAPI
EDUInternalAPI eduServerAPI.EDUServerInputAPI
FederationSenderAPI federationSenderAPI.FederationSenderInternalAPI
FederationSenderAPI federationSenderAPI.FederationInternalAPI
RoomserverAPI roomserverAPI.RoomserverInternalAPI
ServerKeyAPI serverKeyAPI.SigningKeyServerAPI
UserAPI userapi.UserInternalAPI

10
setup/mscs/msc2836/msc2836.go
View file

@ -28,7 +28,7 @@ import (
"time"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
fs "github.com/matrix-org/dendrite/federationsender/api"
fs "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/hooks"
"github.com/matrix-org/dendrite/internal/httputil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
@ -93,7 +93,7 @@ func toClientResponse(res *gomatrixserverlib.MSC2836EventRelationshipsResponse)
// Enable this MSC
func Enable(
base *setup.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI,
base *setup.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
userAPI userapi.UserInternalAPI, keyRing gomatrixserverlib.JSONVerifier,
) error {
db, err := NewDatabase(&base.Cfg.MSCs.Database)
@ -148,10 +148,10 @@ type reqCtx struct {
// federated request args
isFederatedRequest bool
serverName gomatrixserverlib.ServerName
fsAPI fs.FederationSenderInternalAPI
fsAPI fs.FederationInternalAPI
}
func eventRelationshipHandler(db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI) func(*http.Request, *userapi.Device) util.JSONResponse {
func eventRelationshipHandler(db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI) func(*http.Request, *userapi.Device) util.JSONResponse {
return func(req *http.Request, device *userapi.Device) util.JSONResponse {
relation, err := NewEventRelationshipRequest(req.Body)
if err != nil {
@ -183,7 +183,7 @@ func eventRelationshipHandler(db Database, rsAPI roomserver.RoomserverInternalAP
}
func federatedEventRelationship(
ctx context.Context, fedReq *gomatrixserverlib.FederationRequest, db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI,
ctx context.Context, fedReq *gomatrixserverlib.FederationRequest, db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
) util.JSONResponse {
relation, err := NewEventRelationshipRequest(bytes.NewBuffer(fedReq.Content()))
if err != nil {

10
setup/mscs/msc2946/msc2946.go
View file

@ -27,7 +27,7 @@ import (
"github.com/gorilla/mux"
chttputil "github.com/matrix-org/dendrite/clientapi/httputil"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
fs "github.com/matrix-org/dendrite/federationsender/api"
fs "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/hooks"
"github.com/matrix-org/dendrite/internal/httputil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
@ -53,7 +53,7 @@ func Defaults(r *gomatrixserverlib.MSC2946SpacesRequest) {
// Enable this MSC
func Enable(
base *setup.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, userAPI userapi.UserInternalAPI,
fsAPI fs.FederationSenderInternalAPI, keyRing gomatrixserverlib.JSONVerifier,
fsAPI fs.FederationInternalAPI, keyRing gomatrixserverlib.JSONVerifier,
) error {
db, err := NewDatabase(&base.Cfg.MSCs.Database)
if err != nil {
@ -96,7 +96,7 @@ func Enable(
func federatedSpacesHandler(
ctx context.Context, fedReq *gomatrixserverlib.FederationRequest, roomID string, db Database,
rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI,
rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
thisServer gomatrixserverlib.ServerName,
) util.JSONResponse {
inMemoryBatchCache := make(map[string]set)
@ -128,7 +128,7 @@ func federatedSpacesHandler(
}
func spacesHandler(
db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationSenderInternalAPI,
db Database, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
thisServer gomatrixserverlib.ServerName,
) func(*http.Request, *userapi.Device) util.JSONResponse {
return func(req *http.Request, device *userapi.Device) util.JSONResponse {
@ -172,7 +172,7 @@ type walker struct {
thisServer gomatrixserverlib.ServerName
db Database
rsAPI roomserver.RoomserverInternalAPI
fsAPI fs.FederationSenderInternalAPI
fsAPI fs.FederationInternalAPI
ctx context.Context
// user ID|device ID|batch_num => event/room IDs sent to client