keyserver: produce key change events (#1218)

* Produce kafka events when keys are added

* Consume key changes in syncapi with TODO markers for handling them and catching up

* unbreak tests

* Linting

1 files changed, 10 insertions, 1 deletions

diff --git a/keyserver/keyserver.go b/keyserver/keyserver.go
index 2e1ddb6c..47c6a8c3 100644
--- a/keyserver/keyserver.go
+++ b/keyserver/keyserver.go
@@ -15,11 +15,13 @@
 package keyserver
 
 import (
+	"github.com/Shopify/sarama"
 	"github.com/gorilla/mux"
 	"github.com/matrix-org/dendrite/internal/config"
 	"github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/dendrite/keyserver/internal"
 	"github.com/matrix-org/dendrite/keyserver/inthttp"
+	"github.com/matrix-org/dendrite/keyserver/producers"
 	"github.com/matrix-org/dendrite/keyserver/storage"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -34,7 +36,9 @@ func AddInternalRoutes(router *mux.Router, intAPI api.KeyInternalAPI) {
 
 // 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(cfg *config.Dendrite, fedClient *gomatrixserverlib.FederationClient, userAPI userapi.UserInternalAPI) api.KeyInternalAPI {
+func NewInternalAPI(
+	cfg *config.Dendrite, fedClient *gomatrixserverlib.FederationClient, userAPI userapi.UserInternalAPI, producer sarama.SyncProducer,
+) api.KeyInternalAPI {
 	db, err := storage.NewDatabase(
 		string(cfg.Database.E2EKey),
 		cfg.DbProperties(),
@@ -42,10 +46,15 @@ func NewInternalAPI(cfg *config.Dendrite, fedClient *gomatrixserverlib.Federatio
 	if err != nil {
 		logrus.WithError(err).Panicf("failed to connect to key server database")
 	}
+	keyChangeProducer := &producers.KeyChange{
+		Topic:    string(cfg.Kafka.Topics.OutputKeyChangeEvent),
+		Producer: producer,
+	}
 	return &internal.KeyInternalAPI{
 		DB:         db,
 		ThisServer: cfg.Matrix.ServerName,
 		FedClient:  fedClient,
 		UserAPI:    userAPI,
+		Producer:   keyChangeProducer,
 	}
 }