User Postgres database rather than Memory for Naffka

src/github.com/matrix-org/dendrite/cmd/dendrite-monolith-server/main.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"context"
+	"database/sql"
 	"flag"
 	"net/http"
 	"os"
@@ -199,7 +200,21 @@ func (m *monolith) setupFederation() {
 
 func (m *monolith) setupKafka() {
 	if m.cfg.Kafka.UseNaffka {
-		naff, err := naffka.New(&naffka.MemoryDatabase{})
+		db, err := sql.Open("postgres", string(m.cfg.Database.Naffka))
+		if err != nil {
+			log.WithFields(log.Fields{
+				log.ErrorKey: err,
+			}).Panic("Failed to open naffka database")
+		}
+
+		naffkaDB, err := naffka.NewPostgresqlDatabase(db)
+		if err != nil {
+			log.WithFields(log.Fields{
+				log.ErrorKey: err,
+			}).Panic("Failed to setup naffka database")
+		}
+
+		naff, err := naffka.New(naffkaDB)
 		if err != nil {
 			log.WithFields(log.Fields{
 				log.ErrorKey: err,

src/github.com/matrix-org/dendrite/common/config/config.go

@@ -25,8 +25,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/sirupsen/logrus"
 	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
 	"golang.org/x/crypto/ed25519"
 	"gopkg.in/yaml.v2"
 
@@ -148,6 +148,8 @@ type Dendrite struct {
 		// The PublicRoomsAPI database stores information used to compute the public
 		// room directory. It is only accessed by the PublicRoomsAPI server.
 		PublicRoomsAPI DataSource `yaml:"public_rooms_api"`
+		// The Naffka database is used internally by the naffka library, if used.
+		Naffka DataSource `yaml:"naffka,omitempty"`
 	} `yaml:"database"`
 
 	// TURN Server Config
@@ -386,6 +388,8 @@ func (config *Dendrite) check(monolithic bool) error {
 		if !monolithic {
 			problems = append(problems, fmt.Sprintf("naffka can only be used in a monolithic server"))
 		}
+
+		checkNotEmpty("database.naffka", string(config.Database.Naffka))
 	} else {
 		// If we aren't using naffka then we need to have at least one kafka
 		// server to talk to.