Remove eduserver (#2306)

* Move receipt sending to own JetStream producer

* Move SendToDevice to producer

* Remove most parts of the EDU server

* Fix SendToDevice & copyrights

* Move structs, cleanup EDU Server traces

* Use HeadersOnly subscription

* Missing file

* Fix linter issues

* Move consumers to own files

* Rename durable consumer; Consumer cleanup

* Docs/config cleanup

1 files changed, 32 insertions, 9 deletions

diff --git a/federationapi/federationapi.go b/federationapi/federationapi.go
index b7f93ecb..8a0ce8e3 100644
--- a/federationapi/federationapi.go
+++ b/federationapi/federationapi.go
@@ -16,12 +16,12 @@ 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"
 	"github.com/matrix-org/dendrite/federationapi/consumers"
 	"github.com/matrix-org/dendrite/federationapi/internal"
 	"github.com/matrix-org/dendrite/federationapi/inthttp"
+	"github.com/matrix-org/dendrite/federationapi/producers"
 	"github.com/matrix-org/dendrite/federationapi/queue"
 	"github.com/matrix-org/dendrite/federationapi/statistics"
 	"github.com/matrix-org/dendrite/federationapi/storage"
@@ -31,6 +31,7 @@ import (
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/dendrite/setup/process"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/sirupsen/logrus"
 
@@ -46,6 +47,7 @@ func AddInternalRoutes(router *mux.Router, intAPI api.FederationInternalAPI) {
 
 // AddPublicRoutes sets up and registers HTTP handlers on the base API muxes for the FederationAPI component.
 func AddPublicRoutes(
+	process *process.ProcessContext,
 	fedRouter, keyRouter, wellKnownRouter *mux.Router,
 	cfg *config.FederationAPI,
 	userAPI userapi.UserInternalAPI,
@@ -53,16 +55,26 @@ func AddPublicRoutes(
 	keyRing gomatrixserverlib.JSONVerifier,
 	rsAPI roomserverAPI.RoomserverInternalAPI,
 	federationAPI federationAPI.FederationInternalAPI,
-	eduAPI eduserverAPI.EDUServerInputAPI,
 	keyAPI keyserverAPI.KeyInternalAPI,
 	mscCfg *config.MSCs,
 	servers federationAPI.ServersInRoomProvider,
 ) {
+
+	js, _ := jetstream.Prepare(process, &cfg.Matrix.JetStream)
+	producer := &producers.SyncAPIProducer{
+		JetStream:              js,
+		TopicReceiptEvent:      cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
+		TopicSendToDeviceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
+		TopicTypingEvent:       cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
+		ServerName:             cfg.Matrix.ServerName,
+		UserAPI:                userAPI,
+	}
+
 	routing.Setup(
 		fedRouter, keyRouter, wellKnownRouter, cfg, rsAPI,
-		eduAPI, federationAPI, keyRing,
+		federationAPI, keyRing,
 		federation, userAPI, keyAPI, mscCfg,
-		servers,
+		servers, producer,
 	)
 }
 
@@ -112,17 +124,28 @@ func NewInternalAPI(
 	if err = rsConsumer.Start(); err != nil {
 		logrus.WithError(err).Panic("failed to start room server consumer")
 	}
-
-	tsConsumer := consumers.NewOutputEDUConsumer(
+	tsConsumer := consumers.NewOutputSendToDeviceConsumer(
+		base.ProcessContext, cfg, js, queues, federationDB,
+	)
+	if err = tsConsumer.Start(); err != nil {
+		logrus.WithError(err).Panic("failed to start send-to-device consumer")
+	}
+	receiptConsumer := consumers.NewOutputReceiptConsumer(
+		base.ProcessContext, cfg, js, queues, federationDB,
+	)
+	if err = receiptConsumer.Start(); err != nil {
+		logrus.WithError(err).Panic("failed to start receipt consumer")
+	}
+	typingConsumer := consumers.NewOutputTypingConsumer(
 		base.ProcessContext, cfg, js, queues, federationDB,
 	)
-	if err := tsConsumer.Start(); err != nil {
-		logrus.WithError(err).Panic("failed to start typing server consumer")
+	if err = typingConsumer.Start(); err != nil {
+		logrus.WithError(err).Panic("failed to start typing consumer")
 	}
 	keyConsumer := consumers.NewKeyChangeConsumer(
 		base.ProcessContext, &base.Cfg.KeyServer, js, queues, federationDB, rsAPI,
 	)
-	if err := keyConsumer.Start(); err != nil {
+	if err = keyConsumer.Start(); err != nil {
 		logrus.WithError(err).Panic("failed to start key server consumer")
 	}