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

7 files changed, 33 insertions, 58 deletions

diff --git a/setup/base/base.go b/setup/base/base.go
index 6135e080..43d613b0 100644
--- a/setup/base/base.go
+++ b/setup/base/base.go
@@ -45,8 +45,6 @@ import (
 
 	appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
 	asinthttp "github.com/matrix-org/dendrite/appservice/inthttp"
-	eduServerAPI "github.com/matrix-org/dendrite/eduserver/api"
-	eduinthttp "github.com/matrix-org/dendrite/eduserver/inthttp"
 	federationAPI "github.com/matrix-org/dendrite/federationapi/api"
 	federationIntHTTP "github.com/matrix-org/dendrite/federationapi/inthttp"
 	keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
@@ -247,15 +245,6 @@ func (b *BaseDendrite) UserAPIClient() userapi.UserInternalAPI {
 	return userAPI
 }
 
-// EDUServerClient returns EDUServerInputAPI for hitting the EDU server over HTTP
-func (b *BaseDendrite) EDUServerClient() eduServerAPI.EDUServerInputAPI {
-	e, err := eduinthttp.NewEDUServerClient(b.Cfg.EDUServerURL(), b.apiHttpClient)
-	if err != nil {
-		logrus.WithError(err).Panic("EDUServerClient failed", b.apiHttpClient)
-	}
-	return e
-}
-
 // FederationAPIHTTPClient returns FederationInternalAPI for hitting
 // the federation API server over HTTP
 func (b *BaseDendrite) FederationAPIHTTPClient() federationAPI.FederationInternalAPI {
diff --git a/setup/config/config.go b/setup/config/config.go
index eb371a54..e03518e2 100644
--- a/setup/config/config.go
+++ b/setup/config/config.go
@@ -56,7 +56,6 @@ type Dendrite struct {
 	Global        Global        `yaml:"global"`
 	AppServiceAPI AppServiceAPI `yaml:"app_service_api"`
 	ClientAPI     ClientAPI     `yaml:"client_api"`
-	EDUServer     EDUServer     `yaml:"edu_server"`
 	FederationAPI FederationAPI `yaml:"federation_api"`
 	KeyServer     KeyServer     `yaml:"key_server"`
 	MediaAPI      MediaAPI      `yaml:"media_api"`
@@ -296,7 +295,6 @@ func (c *Dendrite) Defaults(generate bool) {
 
 	c.Global.Defaults(generate)
 	c.ClientAPI.Defaults(generate)
-	c.EDUServer.Defaults(generate)
 	c.FederationAPI.Defaults(generate)
 	c.KeyServer.Defaults(generate)
 	c.MediaAPI.Defaults(generate)
@@ -314,8 +312,7 @@ func (c *Dendrite) Verify(configErrs *ConfigErrors, isMonolith bool) {
 		Verify(configErrs *ConfigErrors, isMonolith bool)
 	}
 	for _, c := range []verifiable{
-		&c.Global, &c.ClientAPI,
-		&c.EDUServer, &c.FederationAPI,
+		&c.Global, &c.ClientAPI, &c.FederationAPI,
 		&c.KeyServer, &c.MediaAPI, &c.RoomServer,
 		&c.SyncAPI, &c.UserAPI,
 		&c.AppServiceAPI, &c.MSCs,
@@ -327,7 +324,6 @@ func (c *Dendrite) Verify(configErrs *ConfigErrors, isMonolith bool) {
 func (c *Dendrite) Wiring() {
 	c.Global.JetStream.Matrix = &c.Global
 	c.ClientAPI.Matrix = &c.Global
-	c.EDUServer.Matrix = &c.Global
 	c.FederationAPI.Matrix = &c.Global
 	c.KeyServer.Matrix = &c.Global
 	c.MediaAPI.Matrix = &c.Global
@@ -519,15 +515,6 @@ func (config *Dendrite) UserAPIURL() string {
 	return string(config.UserAPI.InternalAPI.Connect)
 }
 
-// EDUServerURL returns an HTTP URL for where the EDU server is listening.
-func (config *Dendrite) EDUServerURL() string {
-	// Hard code the EDU server to talk HTTP for now.
-	// If we support HTTPS we need to think of a practical way to do certificate validation.
-	// People setting up servers shouldn't need to get a certificate valid for the public
-	// internet for an internal API.
-	return string(config.EDUServer.InternalAPI.Connect)
-}
-
 // KeyServerURL returns an HTTP URL for where the key server is listening.
 func (config *Dendrite) KeyServerURL() string {
 	// Hard code the key server to talk HTTP for now.
diff --git a/setup/config/config_eduserver.go b/setup/config/config_eduserver.go
deleted file mode 100644
index e7ed36aa..00000000
--- a/setup/config/config_eduserver.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package config
-
-type EDUServer struct {
-	Matrix *Global `yaml:"-"`
-
-	InternalAPI InternalAPIOptions `yaml:"internal_api"`
-}
-
-func (c *EDUServer) Defaults(generate bool) {
-	c.InternalAPI.Listen = "http://localhost:7778"
-	c.InternalAPI.Connect = "http://localhost:7778"
-}
-
-func (c *EDUServer) Verify(configErrs *ConfigErrors, isMonolith bool) {
-	checkURL(configErrs, "edu_server.internal_api.listen", string(c.InternalAPI.Listen))
-	checkURL(configErrs, "edu_server.internal_api.connect", string(c.InternalAPI.Connect))
-}
diff --git a/setup/config/config_test.go b/setup/config/config_test.go
index e6f0a493..46e973fa 100644
--- a/setup/config/config_test.go
+++ b/setup/config/config_test.go
@@ -101,10 +101,6 @@ current_state_server:
     max_open_conns: 100
     max_idle_conns: 2
     conn_max_lifetime: -1
-edu_server:
-  internal_api:
-    listen: http://localhost:7778
-    connect: http://localhost:7778
 federation_api:
   internal_api:
     listen: http://localhost:7772
diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go
index 328cf915..4e4fe7a2 100644
--- a/setup/jetstream/nats.go
+++ b/setup/jetstream/nats.go
@@ -157,5 +157,26 @@ func setupNATS(process *process.ProcessContext, cfg *config.JetStream, nc *natsc
 		}
 	}
 
+	// Clean up old consumers so that interest-based consumers do the
+	// right thing.
+	for stream, consumers := range map[string][]string{
+		OutputClientData:        {"SyncAPIClientAPIConsumer"},
+		OutputReceiptEvent:      {"SyncAPIEDUServerReceiptConsumer", "FederationAPIEDUServerConsumer"},
+		OutputSendToDeviceEvent: {"SyncAPIEDUServerSendToDeviceConsumer", "FederationAPIEDUServerConsumer"},
+		OutputTypingEvent:       {"SyncAPIEDUServerTypingConsumer", "FederationAPIEDUServerConsumer"},
+	} {
+		streamName := cfg.Matrix.JetStream.Prefixed(stream)
+		for _, consumer := range consumers {
+			consumerName := cfg.Matrix.JetStream.Prefixed(consumer) + "Pull"
+			consumerInfo, err := s.ConsumerInfo(streamName, consumerName)
+			if err != nil || consumerInfo == nil {
+				continue
+			}
+			if err = s.DeleteConsumer(streamName, consumerName); err != nil {
+				logrus.WithError(err).Errorf("Unable to clean up old consumer %q for stream %q", consumer, stream)
+			}
+		}
+	}
+
 	return s, nc
 }
diff --git a/setup/jetstream/streams.go b/setup/jetstream/streams.go
index aa979924..5f0d37fd 100644
--- a/setup/jetstream/streams.go
+++ b/setup/jetstream/streams.go
@@ -9,8 +9,9 @@ import (
 )
 
 const (
-	UserID = "user_id"
-	RoomID = "room_id"
+	UserID  = "user_id"
+	RoomID  = "room_id"
+	EventID = "event_id"
 )
 
 var (
diff --git a/setup/monolith.go b/setup/monolith.go
index cf6872f9..32f1a649 100644
--- a/setup/monolith.go
+++ b/setup/monolith.go
@@ -19,7 +19,6 @@ import (
 	appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
 	"github.com/matrix-org/dendrite/clientapi"
 	"github.com/matrix-org/dendrite/clientapi/api"
-	eduServerAPI "github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/federationapi"
 	federationAPI "github.com/matrix-org/dendrite/federationapi/api"
 	"github.com/matrix-org/dendrite/internal/transactions"
@@ -43,12 +42,11 @@ type Monolith struct {
 	Client    *gomatrixserverlib.Client
 	FedClient *gomatrixserverlib.FederationClient
 
-	AppserviceAPI  appserviceAPI.AppServiceQueryAPI
-	EDUInternalAPI eduServerAPI.EDUServerInputAPI
-	FederationAPI  federationAPI.FederationInternalAPI
-	RoomserverAPI  roomserverAPI.RoomserverInternalAPI
-	UserAPI        userapi.UserInternalAPI
-	KeyAPI         keyAPI.KeyInternalAPI
+	AppserviceAPI appserviceAPI.AppServiceQueryAPI
+	FederationAPI federationAPI.FederationInternalAPI
+	RoomserverAPI roomserverAPI.RoomserverInternalAPI
+	UserAPI       userapi.UserInternalAPI
+	KeyAPI        keyAPI.KeyInternalAPI
 
 	// Optional
 	ExtPublicRoomsProvider   api.ExtraPublicRoomsProvider
@@ -64,14 +62,14 @@ func (m *Monolith) AddAllPublicRoutes(process *process.ProcessContext, csMux, ss
 	clientapi.AddPublicRoutes(
 		process, csMux, synapseMux, &m.Config.ClientAPI,
 		m.FedClient, m.RoomserverAPI,
-		m.EDUInternalAPI, m.AppserviceAPI, transactions.New(),
+		m.AppserviceAPI, transactions.New(),
 		m.FederationAPI, m.UserAPI, userDirectoryProvider, m.KeyAPI,
 		m.ExtPublicRoomsProvider, &m.Config.MSCs,
 	)
 	federationapi.AddPublicRoutes(
-		ssMux, keyMux, wkMux, &m.Config.FederationAPI, m.UserAPI, m.FedClient,
+		process, ssMux, keyMux, wkMux, &m.Config.FederationAPI, m.UserAPI, m.FedClient,
 		m.KeyRing, m.RoomserverAPI, m.FederationAPI,
-		m.EDUInternalAPI, m.KeyAPI, &m.Config.MSCs, nil,
+		m.KeyAPI, &m.Config.MSCs, nil,
 	)
 	mediaapi.AddPublicRoutes(mediaMux, &m.Config.MediaAPI, &m.Config.ClientAPI.RateLimiting, m.UserAPI, m.Client)
 	syncapi.AddPublicRoutes(