Update Naffka (#1295)

* Update Naffka

* Fix Naffka setup

* Update Naffka

* Update Naffka

currentstateserver/currentstateserver_test.go

@@ -34,11 +34,11 @@ import (
 	"github.com/matrix-org/dendrite/currentstateserver/storage"
 	"github.com/matrix-org/dendrite/internal/config"
 	"github.com/matrix-org/dendrite/internal/httputil"
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/internal/test"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/naffka"
+	naffkaStorage "github.com/matrix-org/naffka/storage"
 )
 
 var (
@@ -122,13 +122,7 @@ func MustMakeInternalAPI(t *testing.T) (api.CurrentStateInternalAPI, storage.Dat
 	cfg.Global.ServerName = "kaer.morhen"
 	cfg.CurrentStateServer.Database.ConnectionString = config.DataSource("file:" + stateDBName)
 	cfg.Global.Kafka.TopicPrefix = kafkaPrefix
-	db, err := sqlutil.Open(&config.DatabaseOptions{
+	naffkaDB, err := naffkaStorage.NewDatabase("file:" + naffkaDBName)
-		ConnectionString: config.DataSource("file:" + naffkaDBName),
-	})
-	if err != nil {
-		t.Fatalf("Failed to open naffka database: %s", err)
-	}
-	naffkaDB, err := naffka.NewSqliteDatabase(db)
 	if err != nil {
 		t.Fatalf("Failed to setup naffka database: %s", err)
 	}

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3
 	github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26
 	github.com/matrix-org/gomatrixserverlib v0.0.0-20200817100842-9d02141812f2
-	github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f
+	github.com/matrix-org/naffka v0.0.0-20200824124823-ed1d3c8c35f5
 	github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
 	github.com/mattn/go-sqlite3 v2.0.2+incompatible
 	github.com/nfnt/resize v0.0.0-20160724205520-891127d8d1b5

go.sum

@@ -149,8 +149,6 @@ github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
 github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gorilla/mux v1.7.3 h1:gnP5JzjVOuiZD07fKKToCAOjS0yOpj/qPETTXCCS6hw=
-github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
 github.com/gorilla/mux v1.7.4 h1:VuZ8uybHlWmqV03+zRzdwKL4tUnIp1MAQtp1mIFE1bc=
 github.com/gorilla/mux v1.7.4/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM=
@@ -426,8 +424,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh
 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20200817100842-9d02141812f2 h1:9wKwfd5KDcXuqZ7/kAaYe0QM4DGM+2awjjvXQtrDa6k=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20200817100842-9d02141812f2/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
-github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f h1:pRz4VTiRCO4zPlEMc3ESdUOcW4PXHH4Kj+YDz1XyE+Y=
+github.com/matrix-org/naffka v0.0.0-20200824124823-ed1d3c8c35f5 h1:1F49phdXDbU+wKnSpwf0IrfAaLnDCFLIloPLh8zupA8=
-github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f/go.mod h1:y0oDTjZDv5SM9a2rp3bl+CU+bvTRINQsdb7YlDql5Go=
+github.com/matrix-org/naffka v0.0.0-20200824124823-ed1d3c8c35f5/go.mod h1:O4o8X87YrFWi+FKvBqS1wuT6W/parw2BlzveXu1sHyY=
 github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 h1:ntrLa/8xVzeSs8vHFHK25k0C+NV74sYMJnNSg5NoSRo=
 github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U=
 github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4 h1:eCEHXWDv9Rm335MSuB49mFUK44bwZPFSDde3ORE3syk=

internal/setup/base.go

@@ -24,11 +24,12 @@ import (
 	currentstateAPI "github.com/matrix-org/dendrite/currentstateserver/api"
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/httputil"
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/matrix-org/naffka"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
+	"github.com/matrix-org/naffka"
+	naffkaStorage "github.com/matrix-org/naffka/storage"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/userapi/storage/accounts"
 	"github.com/matrix-org/dendrite/userapi/storage/devices"
@@ -356,36 +357,13 @@ func setupKafka(cfg *config.Dendrite) (sarama.Consumer, sarama.SyncProducer) {
 
 // setupNaffka creates kafka consumer/producer pair from the config.
 func setupNaffka(cfg *config.Dendrite) (sarama.Consumer, sarama.SyncProducer) {
-	var naffkaDB *naffka.DatabaseImpl
+	naffkaDB, err := naffkaStorage.NewDatabase(string(cfg.Global.Kafka.Database.ConnectionString))
-
-	db, err := sqlutil.Open(&cfg.Global.Kafka.Database)
-	if err != nil {
-		logrus.WithError(err).Panic("Failed to open naffka database")
-	}
-
-	switch {
-	case cfg.Global.Kafka.Database.ConnectionString.IsSQLite():
-		naffkaDB, err = naffka.NewSqliteDatabase(db)
 	if err != nil {
 		logrus.WithError(err).Panic("Failed to setup naffka database")
 	}
-	case cfg.Global.Kafka.Database.ConnectionString.IsPostgres():
-		naffkaDB, err = naffka.NewPostgresqlDatabase(db)
-		if err != nil {
-			logrus.WithError(err).Panic("Failed to setup naffka database")
-		}
-	default:
-		panic("unknown naffka database type")
-	}
-
-	if naffkaDB == nil {
-		panic("naffka connection string not understood")
-	}
-
 	naff, err := naffka.New(naffkaDB)
 	if err != nil {
 		logrus.WithError(err).Panic("Failed to setup naffka")
 	}
-
 	return naff, naff
 }