aboutsummaryrefslogtreecommitdiff
path: root/cmd/dendrite-demo-pinecone/main.go
diff options
context:
space:
mode:
Diffstat (limited to 'cmd/dendrite-demo-pinecone/main.go')
-rw-r--r--cmd/dendrite-demo-pinecone/main.go125
1 files changed, 117 insertions, 8 deletions
diff --git a/cmd/dendrite-demo-pinecone/main.go b/cmd/dendrite-demo-pinecone/main.go
index 3f627b41..a813c37a 100644
--- a/cmd/dendrite-demo-pinecone/main.go
+++ b/cmd/dendrite-demo-pinecone/main.go
@@ -38,16 +38,21 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/api"
+ "github.com/matrix-org/dendrite/federationapi/producers"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/keyserver"
+ "github.com/matrix-org/dendrite/relayapi"
+ relayServerAPI "github.com/matrix-org/dendrite/relayapi/api"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/userapi"
"github.com/matrix-org/gomatrixserverlib"
+ "go.uber.org/atomic"
pineconeConnections "github.com/matrix-org/pinecone/connections"
pineconeMulticast "github.com/matrix-org/pinecone/multicast"
@@ -66,6 +71,8 @@ var (
instanceDir = flag.String("dir", ".", "the directory to store the databases in (if --config not specified)")
)
+const relayServerRetryInterval = time.Second * 30
+
// nolint:gocyclo
func main() {
flag.Parse()
@@ -139,6 +146,7 @@ func main() {
cfg.RoomServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-roomserver.db", filepath.Join(*instanceDir, *instanceName)))
cfg.KeyServer.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-keyserver.db", filepath.Join(*instanceDir, *instanceName)))
cfg.FederationAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-federationapi.db", filepath.Join(*instanceDir, *instanceName)))
+ cfg.RelayAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-relayapi.db", filepath.Join(*instanceDir, *instanceName)))
cfg.MSCs.MSCs = []string{"msc2836", "msc2946"}
cfg.MSCs.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mscs.db", filepath.Join(*instanceDir, *instanceName)))
cfg.ClientAPI.RegistrationDisabled = false
@@ -224,6 +232,20 @@ func main() {
userProvider := users.NewPineconeUserProvider(pRouter, pQUIC, userAPI, federation)
roomProvider := rooms.NewPineconeRoomProvider(pRouter, pQUIC, fsAPI, federation)
+ js, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
+ producer := &producers.SyncAPIProducer{
+ JetStream: js,
+ TopicReceiptEvent: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputReceiptEvent),
+ TopicSendToDeviceEvent: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
+ TopicTypingEvent: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputTypingEvent),
+ TopicPresenceEvent: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
+ TopicDeviceListUpdate: base.Cfg.Global.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
+ TopicSigningKeyUpdate: base.Cfg.Global.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
+ Config: &base.Cfg.FederationAPI,
+ UserAPI: userAPI,
+ }
+ relayAPI := relayapi.NewRelayInternalAPI(base, federation, rsAPI, keyRing, producer)
+
monolith := setup.Monolith{
Config: base.Cfg,
Client: conn.CreateClient(base, pQUIC),
@@ -235,6 +257,7 @@ func main() {
RoomserverAPI: rsAPI,
UserAPI: userAPI,
KeyAPI: keyAPI,
+ RelayAPI: relayAPI,
ExtPublicRoomsProvider: roomProvider,
ExtUserDirectoryProvider: userProvider,
}
@@ -305,27 +328,38 @@ func main() {
go func(ch <-chan pineconeEvents.Event) {
eLog := logrus.WithField("pinecone", "events")
+ relayServerSyncRunning := atomic.NewBool(false)
+ stopRelayServerSync := make(chan bool)
+
+ m := RelayServerRetriever{
+ Context: context.Background(),
+ ServerName: gomatrixserverlib.ServerName(pRouter.PublicKey().String()),
+ FederationAPI: fsAPI,
+ RelayServersQueried: make(map[gomatrixserverlib.ServerName]bool),
+ RelayAPI: monolith.RelayAPI,
+ }
+ m.InitializeRelayServers(eLog)
for event := range ch {
switch e := event.(type) {
case pineconeEvents.PeerAdded:
+ if !relayServerSyncRunning.Load() {
+ go m.syncRelayServers(stopRelayServerSync, *relayServerSyncRunning)
+ }
case pineconeEvents.PeerRemoved:
- case pineconeEvents.TreeParentUpdate:
- case pineconeEvents.SnakeDescUpdate:
- case pineconeEvents.TreeRootAnnUpdate:
- case pineconeEvents.SnakeEntryAdded:
- case pineconeEvents.SnakeEntryRemoved:
+ if relayServerSyncRunning.Load() && pRouter.TotalPeerCount() == 0 {
+ stopRelayServerSync <- true
+ }
case pineconeEvents.BroadcastReceived:
- eLog.Info("Broadcast received from: ", e.PeerID)
+ // eLog.Info("Broadcast received from: ", e.PeerID)
req := &api.PerformWakeupServersRequest{
ServerNames: []gomatrixserverlib.ServerName{gomatrixserverlib.ServerName(e.PeerID)},
}
res := &api.PerformWakeupServersResponse{}
if err := fsAPI.PerformWakeupServers(base.Context(), req, res); err != nil {
- logrus.WithError(err).Error("Failed to wakeup destination", e.PeerID)
+ eLog.WithError(err).Error("Failed to wakeup destination", e.PeerID)
}
- case pineconeEvents.BandwidthReport:
default:
}
}
@@ -333,3 +367,78 @@ func main() {
base.WaitForShutdown()
}
+
+type RelayServerRetriever struct {
+ Context context.Context
+ ServerName gomatrixserverlib.ServerName
+ FederationAPI api.FederationInternalAPI
+ RelayServersQueried map[gomatrixserverlib.ServerName]bool
+ RelayAPI relayServerAPI.RelayInternalAPI
+}
+
+func (m *RelayServerRetriever) InitializeRelayServers(eLog *logrus.Entry) {
+ request := api.P2PQueryRelayServersRequest{Server: gomatrixserverlib.ServerName(m.ServerName)}
+ response := api.P2PQueryRelayServersResponse{}
+ err := m.FederationAPI.P2PQueryRelayServers(m.Context, &request, &response)
+ if err != nil {
+ eLog.Warnf("Failed obtaining list of this node's relay servers: %s", err.Error())
+ }
+ for _, server := range response.RelayServers {
+ m.RelayServersQueried[server] = false
+ }
+
+ eLog.Infof("Registered relay servers: %v", response.RelayServers)
+}
+
+func (m *RelayServerRetriever) syncRelayServers(stop <-chan bool, running atomic.Bool) {
+ defer running.Store(false)
+
+ t := time.NewTimer(relayServerRetryInterval)
+ for {
+ relayServersToQuery := []gomatrixserverlib.ServerName{}
+ for server, complete := range m.RelayServersQueried {
+ if !complete {
+ relayServersToQuery = append(relayServersToQuery, server)
+ }
+ }
+ if len(relayServersToQuery) == 0 {
+ // All relay servers have been synced.
+ return
+ }
+ m.queryRelayServers(relayServersToQuery)
+ t.Reset(relayServerRetryInterval)
+
+ select {
+ case <-stop:
+ // We have been asked to stop syncing, drain the timer and return.
+ if !t.Stop() {
+ <-t.C
+ }
+ return
+ case <-t.C:
+ // The timer has expired. Continue to the next loop iteration.
+ }
+ }
+}
+
+func (m *RelayServerRetriever) queryRelayServers(relayServers []gomatrixserverlib.ServerName) {
+ logrus.Info("querying relay servers for any available transactions")
+ for _, server := range relayServers {
+ userID, err := gomatrixserverlib.NewUserID("@user:"+string(m.ServerName), false)
+ if err != nil {
+ return
+ }
+ err = m.RelayAPI.PerformRelayServerSync(context.Background(), *userID, server)
+ if err == nil {
+ m.RelayServersQueried[server] = true
+ // TODO : What happens if your relay receives new messages after this point?
+ // Should you continue to check with them, or should they try and contact you?
+ // They could send a "new_async_events" message your way maybe?
+ // Then you could mark them as needing to be queried again.
+ // What if you miss this message?
+ // Maybe you should try querying them again after a certain period of time as a backup?
+ } else {
+ logrus.Errorf("Failed querying relay server: %s", err.Error())
+ }
+ }
+}
generated by cgit v1.2.3 (git 2.26.2) at 2025-02-27 19:32:50 +0000