Repository: textileio/go-threads
Pull request #468 (2020-12-04 23:34:07)

go-ds-mongo
Allows the use of [`go-ds-mongo`](https://github.com/textileio/go-ds-mongo) for all needed persistence.


Patch:
diff --git a/.github/workflows/publish-grpc-libs.yml b/.github/workflows/publish-grpc-libs.yml
index a89c0d0..51ddedf 100644
--- a/.github/workflows/publish-grpc-libs.yml
+++ b/.github/workflows/publish-grpc-libs.yml
@@ -23,7 +23,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v1
         with:
-          go-version: 1.14
+          go-version: 1.15
       - name: Setup env
         env:
           ACTIONS_ALLOW_UNSECURE_COMMANDS: true
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 33e0adc..c17f5c2 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -10,7 +10,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v1
         with:
-          go-version: 1.14
+          go-version: 1.15
       - name: Check out code
         uses: actions/checkout@v1
       - name: Cache dependencies
diff --git a/Dockerfile b/Dockerfile
index 376b8da..ab44d00 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM golang:1.14.2-buster
+FROM golang:1.15.5-buster
 MAINTAINER Textile <contact@textile.io>
 
 # This is (in large part) copied (with love) from
diff --git a/Makefile b/Makefile
index a6275e5..b8af386 100644
--- a/Makefile
+++ b/Makefile
@@ -1,3 +1,12 @@
-docker:
-	$(eval VERSION := $(shell git --no-pager describe --abbrev=0 --tags --always))
-	docker build -t go-threads:$(VERSION:v%=%) .
\ No newline at end of file
+threaddb-up:
+	docker-compose -f docker-compose-dev.yml up --build
+
+threaddb-stop:
+	docker-compose -f docker-compose-dev.yml stop
+
+threaddb-clean:
+	docker-compose -f docker-compose-dev.yml down -v --remove-orphans
+
+test:
+	go test -race -timeout 45m ./...
+.PHONY: test
\ No newline at end of file
diff --git a/api/service.go b/api/service.go
index 9319ef7..d50af0f 100644
--- a/api/service.go
+++ b/api/service.go
@@ -17,6 +17,7 @@ import (
 	lstore "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/core/thread"
 	"github.com/textileio/go-threads/db"
+	kt "github.com/textileio/go-threads/db/keytransform"
 	"github.com/textileio/go-threads/util"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -31,15 +32,14 @@ type Service struct {
 	manager *db.Manager
 }
 
-// Config specifies server settings.
+// Config specifies service settings.
 type Config struct {
-	RepoPath string
-	Debug    bool
+	Debug bool
 }
 
 // NewService starts and returns a new service with the given network.
 // The network is *not* managed by the server.
-func NewService(network app.Net, conf Config) (*Service, error) {
+func NewService(store kt.TxnDatastoreExtended, network app.Net, conf Config) (*Service, error) {
 	var err error
 	if conf.Debug {
 		err = util.SetLogLevels(map[string]logging.LogLevel{
@@ -50,7 +50,7 @@ func NewService(network app.Net, conf Config) (*Service, error) {
 		}
 	}
 
-	manager, err := db.NewManager(network, db.WithNewRepoPath(conf.RepoPath), db.WithNewDebug(conf.Debug))
+	manager, err := db.NewManager(store, network, db.WithNewDebug(conf.Debug))
 	if err != nil {
 		return nil, err
 	}
diff --git a/common/common.go b/common/common.go
index dfc140b..d24c74d 100644
--- a/common/common.go
+++ b/common/common.go
@@ -2,18 +2,23 @@ package common
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io/ioutil"
 	"os"
 	"path/filepath"
 	"time"
 
 	ipfslite "github.com/hsanjuan/ipfs-lite"
+	ds "github.com/ipfs/go-datastore"
 	"github.com/libp2p/go-libp2p"
 	connmgr "github.com/libp2p/go-libp2p-connmgr"
 	cconnmgr "github.com/libp2p/go-libp2p-core/connmgr"
+	"github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-peerstore/pstoreds"
 	ma "github.com/multiformats/go-multiaddr"
+	mongods "github.com/textileio/go-ds-mongo"
 	"github.com/textileio/go-threads/core/app"
 	core "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/logstore/lstoreds"
@@ -24,11 +29,6 @@ import (
 	"google.golang.org/grpc"
 )
 
-const (
-	defaultIpfsLitePath = "ipfslite"
-	defaultLogstorePath = "logstore"
-)
-
 // DefaultNetwork is a boostrapable default Net with sane defaults.
 type NetBoostrapper interface {
 	app.Net
@@ -36,7 +36,7 @@ type NetBoostrapper interface {
 	Bootstrap(addrs []peer.AddrInfo)
 }
 
-func DefaultNetwork(repoPath string, opts ...NetOption) (NetBoostrapper, error) {
+func DefaultNetwork(opts ...NetOption) (NetBoostrapper, error) {
 	var (
 		config NetConfig
 		fin    = util.NewFinalizer()
@@ -52,30 +52,29 @@ func DefaultNetwork(repoPath string, opts ...NetOption) (NetBoostrapper, error)
 		return nil, err
 	}
 
-	ipfsLitePath := filepath.Join(repoPath, defaultIpfsLitePath)
-	if err := os.MkdirAll(ipfsLitePath, os.ModePerm); err != nil {
-		return nil, err
-	}
+	ctx, cancel := context.WithCancel(context.Background())
+	fin.Add(util.NewContextCloser(cancel))
 
-	litestore, err := ipfslite.BadgerDatastore(ipfsLitePath)
+	litestore, err := persistentStore(ctx, config, "ipfslite", fin)
 	if err != nil {
-		return nil, err
+		return nil, fin.Cleanup(err)
 	}
 	fin.Add(litestore)
 
-	ctx, cancel := context.WithCancel(context.Background())
-	fin.Add(util.NewContextCloser(cancel))
-
 	pstore, err := pstoreds.NewPeerstore(ctx, litestore, pstoreds.DefaultOpts())
 	if err != nil {
 		return nil, fin.Cleanup(err)
 	}
 	fin.Add(pstore)
 
-	priv := util.LoadKey(filepath.Join(ipfsLitePath, "key"))
+	hostKey, err := getHostKey(config, litestore)
+	if err != nil {
+		return nil, fin.Cleanup(err)
+	}
+
 	h, d, err := ipfslite.SetupLibp2p(
 		ctx,
-		priv,
+		hostKey,
 		nil,
 		[]ma.Multiaddr{config.HostAddr},
 		litestore,
@@ -92,7 +91,7 @@ func DefaultNetwork(repoPath string, opts ...NetOption) (NetBoostrapper, error)
 		return nil, fin.Cleanup(err)
 	}
 
-	tstore, err := buildLogstore(ctx, config.LSType, repoPath, fin)
+	tstore, err := buildLogstore(ctx, config, fin)
 	if err != nil {
 		return nil, fin.Cleanup(err)
 	}
@@ -114,13 +113,13 @@ func DefaultNetwork(repoPath string, opts ...NetOption) (NetBoostrapper, error)
 	}, nil
 }
 
-func buildLogstore(ctx context.Context, lstype LogstoreType, repoPath string, fin *util.Finalizer) (core.Logstore, error) {
-	switch lstype {
+func buildLogstore(ctx context.Context, config NetConfig, fin *util.Finalizer) (core.Logstore, error) {
+	switch config.LSType {
 	case LogstoreInMemory:
 		return lstoremem.NewLogstore(), nil
 
 	case LogstoreHybrid:
-		pls, err := persistentLogstore(ctx, repoPath, fin)
+		pls, err := persistentLogstore(ctx, config, fin)
 		if err != nil {
 			return nil, err
 		}
@@ -128,26 +127,109 @@ func buildLogstore(ctx context.Context, lstype LogstoreType, repoPath string, fi
 		return lstorehybrid.NewLogstore(pls, mls)
 
 	case LogstorePersistent:
-		return persistentLogstore(ctx, repoPath, fin)
+		return persistentLogstore(ctx, config, fin)
 
 	default:
-		return nil, fmt.Errorf("unsupported logstore type: %s", lstype)
+		return nil, fmt.Errorf("unsupported logstore type: %s", config.LSType)
+	}
+}
+
+func persistentLogstore(ctx context.Context, config NetConfig, fin *util.Finalizer) (core.Logstore, error) {
+	pds, err := persistentStore(ctx, config, "logstore", fin)
+	if err != nil {
+		return nil, err
+	}
+	return lstoreds.NewLogstore(ctx, pds, lstoreds.DefaultOpts())
+}
+
+func persistentStore(ctx context.Context, config NetConfig, name string, fin *util.Finalizer) (ds.Batching, error) {
+	if len(config.MongoUri) != 0 {
+		return mongoStore(ctx, config.MongoUri, config.MongoDB, name, fin)
+	} else {
+		return badgerStore(filepath.Join(config.BadgerRepoPath, name), fin)
 	}
 }
 
-func persistentLogstore(ctx context.Context, repoPath string, fin *util.Finalizer) (core.Logstore, error) {
-	logstorePath := filepath.Join(repoPath, defaultLogstorePath)
-	if err := os.MkdirAll(logstorePath, os.ModePerm); err != nil {
+func badgerStore(repoPath string, fin *util.Finalizer) (ds.Batching, error) {
+	if err := os.MkdirAll(repoPath, os.ModePerm); err != nil {
 		return nil, err
 	}
 
-	dstore, err := ipfslite.BadgerDatastore(logstorePath)
+	dstore, err := ipfslite.BadgerDatastore(repoPath)
+	if err != nil {
+		return nil, err
+	}
+	fin.Add(dstore)
+
+	return dstore, nil
+}
+
+func mongoStore(ctx context.Context, uri, db, collection string, fin *util.Finalizer) (ds.Batching, error) {
+	dstore, err := mongods.New(ctx, uri, db, mongods.WithCollName(collection))
 	if err != nil {
 		return nil, err
 	}
 	fin.Add(dstore)
 
-	return lstoreds.NewLogstore(ctx, dstore, lstoreds.DefaultOpts())
+	return dstore, nil
+}
+
+func getHostKey(config NetConfig, store ds.Batching) (crypto.PrivKey, error) {
+	if len(config.MongoUri) != 0 {
+		k := ds.NewKey("key")
+		bytes, err := store.Get(k)
+		if errors.Is(err, ds.ErrNotFound) {
+			key, bytes, err := newHostKey()
+			if err != nil {
+				return nil, err
+			}
+			if err = store.Put(k, bytes); err != nil {
+				return nil, err
+			}
+			return key, nil
+		} else if err != nil {
+			return nil, err
+		}
+		return crypto.UnmarshalPrivateKey(bytes)
+	} else {
+		// If a local datastore is used, the key is written to a file
+		dir := filepath.Join(config.BadgerRepoPath, "ipfslite")
+		if err := os.MkdirAll(dir, os.ModePerm); err != nil {
+			return nil, err
+		}
+		pth := filepath.Join(dir, "key")
+		_, err := os.Stat(pth)
+		if os.IsNotExist(err) {
+			key, bytes, err := newHostKey()
+			if err != nil {
+				return nil, err
+			}
+			if err = ioutil.WriteFile(pth, bytes, 0400); err != nil {
+				return nil, err
+			}
+			return key, nil
+		} else if err != nil {
+			return nil, err
+		} else {
+			bytes, err := ioutil.ReadFile(pth)
+			if err != nil {
+				return nil, err
+			}
+			return crypto.UnmarshalPrivateKey(bytes)
+		}
+	}
+}
+
+func newHostKey() (crypto.PrivKey, []byte, error) {
+	priv, _, err := crypto.GenerateKeyPair(crypto.Ed25519, 0)
+	if err != nil {
+		return nil, nil, err
+	}
+	key, err := crypto.MarshalPrivateKey(priv)
+	if err != nil {
+		return nil, nil, err
+	}
+	return priv, key, nil
 }
 
 func setDefaults(config *NetConfig) error {
@@ -167,6 +249,10 @@ func setDefaults(config *NetConfig) error {
 		config.LSType = LogstorePersistent
 	}
 
+	if len(config.MongoDB) == 0 {
+		config.MongoDB = "threadnet"
+	}
+
 	return nil
 }
 
@@ -184,6 +270,9 @@ type NetConfig struct {
 	GRPCServerOptions []grpc.ServerOption
 	GRPCDialOptions   []grpc.DialOption
 	LSType            LogstoreType
+	BadgerRepoPath    string
+	MongoUri          string
+	MongoDB           string
 	PubSub            bool
 	Debug             bool
 }
@@ -239,6 +328,21 @@ func WithNetLogstore(lt LogstoreType) NetOption {
 	}
 }
 
+func WithNetBadgerPersistence(repoPath string) NetOption {
+	return func(c *NetConfig) error {
+		c.BadgerRepoPath = repoPath
+		return nil
+	}
+}
+
+func WithNetMongoPersistence(uri, db string) NetOption {
+	return func(c *NetConfig) error {
+		c.MongoUri = uri
+		c.MongoDB = db
+		return nil
+	}
+}
+
 type netBoostrapper struct {
 	app.Net
 	litepeer  *ipfslite.Peer
diff --git a/core/app/app.go b/core/app/app.go
index 9c8cb4e..0942398 100644
--- a/core/app/app.go
+++ b/core/app/app.go
@@ -7,7 +7,6 @@ import (
 	"time"
 
 	format "github.com/ipfs/go-ipld-format"
-	logging "github.com/ipfs/go-log"
 	"github.com/textileio/go-threads/broadcast"
 	"github.com/textileio/go-threads/core/net"
 	"github.com/textileio/go-threads/core/thread"
@@ -15,8 +14,6 @@ import (
 )
 
 var (
-	log = logging.Logger("app")
-
 	// ErrThreadInUse indicates an operation could not be completed because the
 	// thread is bound to an app.
 	ErrThreadInUse = errors.New("thread is in use")
diff --git a/core/db/db.go b/core/db/db.go
index 9d4ca3c..2f85931 100644
--- a/core/db/db.go
+++ b/core/db/db.go
@@ -4,9 +4,9 @@ import (
 	"crypto/rand"
 	"strings"
 
+	ds "github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-ipld-format"
 	ulid "github.com/oklog/ulid/v2"
-	ds "github.com/textileio/go-datastore"
 )
 
 const (
@@ -82,7 +82,7 @@ type IndexFunc func(collection string, key ds.Key, oldData, newData []byte, txn
 // events dispatched to thread logs, and viceversa.
 type EventCodec interface {
 	// Reduce applies generated events into state.
-	Reduce(events []Event, datastore ds.TxnDatastore, baseKey ds.Key, indexFunc IndexFunc) ([]ReduceAction, error)
+	Reduce(events []Event, store ds.TxnDatastore, baseKey ds.Key, indexFunc IndexFunc) ([]ReduceAction, error)
 	// Create corresponding events to be dispatched.
 	Create(ops []Action) ([]Event, format.Node, error)
 	// EventsFromBytes deserializes a format.Node bytes payload into Events.
diff --git a/db/collection.go b/db/collection.go
index d5b651f..096f2ff 100644
--- a/db/collection.go
+++ b/db/collection.go
@@ -13,9 +13,10 @@ import (
 	"github.com/alecthomas/jsonschema"
 	"github.com/dop251/goja"
 	jsonpatch "github.com/evanphx/json-patch"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	format "github.com/ipfs/go-ipld-format"
-	ds "github.com/textileio/go-datastore"
-	"github.com/textileio/go-datastore/query"
+	dse "github.com/textileio/go-datastore-extensions"
 	"github.com/textileio/go-threads/core/app"
 	core "github.com/textileio/go-threads/core/db"
 	"github.com/textileio/go-threads/core/thread"
@@ -39,7 +40,6 @@ var (
 	ErrInvalidSchemaInstance = errors.New("instance doesn't correspond to schema")
 
 	errMissingInstanceID           = errors.New("invalid instance: missing _id attribute")
-	errMissingModTag               = errors.New("invalid instance: missing _mod attribute")
 	errAlreadyDiscardedCommitedTxn = errors.New("can't commit discarded/committed txn")
 	errCantCreateExistingInstance  = errors.New("can't create already existing instance")
 
@@ -290,23 +290,24 @@ func (c *Collection) ModifiedSince(time int64, opts ...TxnOption) (modified []co
 	c.db.dispatcher.Lock().Lock()
 	defer c.db.dispatcher.Lock().Unlock()
 
-	txn, err := c.db.dispatcher.Store().NewTransaction(false)
+	txn, err := c.db.dispatcher.Store().NewTransactionExtended(false)
 	if err != nil {
 		return nil, err
 	}
 	defer txn.Discard()
 
-	query := query.Query{
-		Prefix:     dsDispatcherPrefix.String(),
-		SeekPrefix: dsDispatcherPrefix.ChildString(strconv.FormatInt(time, 10)).String(),
-		Filters: []query.Filter{
-			filter{
-				Collection: c.name,
+	results, err := txn.QueryExtended(dse.QueryExt{
+		Query: query.Query{
+			Prefix: dsDispatcherPrefix.String(),
+			Filters: []query.Filter{
+				filter{
+					Collection: c.name,
+				},
 			},
+			KeysOnly: true,
 		},
-		KeysOnly: true,
-	}
-	results, err := txn.Query(query)
+		SeekPrefix: dsDispatcherPrefix.ChildString(strconv.FormatInt(time, 10)).String(),
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -765,19 +766,6 @@ func setNewInstanceID(t []byte) (core.InstanceID, []byte) {
 	return newID, patchedValue
 }
 
-func getModifiedTag(t []byte) (int64, error) {
-	partial := &struct {
-		Mod *int64 `json:"_mod"`
-	}{}
-	if err := json.Unmarshal(t, partial); err != nil {
-		return 0, fmt.Errorf("unmarshaling json instance: %v", err)
-	}
-	if partial.Mod == nil {
-		return 0, errMissingInstanceID
-	}
-	return *partial.Mod, nil
-}
-
 func setModifiedTag(t []byte) (newTime int64, patchedValue []byte) {
 	newTime = time.Now().UnixNano()
 	patchedValue, err := jsonpatch.MergePatch(t, []byte(fmt.Sprintf(`{"%s": %d}`, modFieldName, newTime)))
diff --git a/db/common.go b/db/common.go
index 962c4f3..74b5d77 100644
--- a/db/common.go
+++ b/db/common.go
@@ -5,28 +5,43 @@ import (
 	"sync"
 	"time"
 
-	datastore "github.com/textileio/go-datastore"
-	"github.com/textileio/go-datastore/query"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
+	dse "github.com/textileio/go-datastore-extensions"
 	core "github.com/textileio/go-threads/core/db"
 )
 
-type TxMapDatastore struct {
-	*datastore.MapDatastore
+type TxnMapDatastore struct {
+	*ds.MapDatastore
 	lock sync.RWMutex
 }
 
-func NewTxMapDatastore() *TxMapDatastore {
-	return &TxMapDatastore{
-		MapDatastore: datastore.NewMapDatastore(),
+var _ dse.DatastoreExtensions = (*TxnMapDatastore)(nil)
+
+func NewTxMapDatastore() *TxnMapDatastore {
+	return &TxnMapDatastore{
+		MapDatastore: ds.NewMapDatastore(),
 	}
 }
 
-func (d *TxMapDatastore) NewTransaction(_ bool) (datastore.Txn, error) {
+func (d *TxnMapDatastore) NewTransaction(_ bool) (ds.Txn, error) {
 	d.lock.RLock()
 	defer d.lock.RUnlock()
 	return NewSimpleTx(d), nil
 }
 
+func (d *TxnMapDatastore) NewTransactionExtended(_ bool) (dse.TxnExt, error) {
+	d.lock.RLock()
+	defer d.lock.RUnlock()
+	return NewSimpleTx(d), nil
+}
+
+func (d *TxnMapDatastore) QueryExtended(q dse.QueryExt) (query.Results, error) {
+	d.lock.RLock()
+	defer d.lock.RUnlock()
+	return d.Query(q.Query)
+}
+
 type op struct {
 	delete bool
 	value  []byte
@@ -35,15 +50,15 @@ type op struct {
 // SimpleTx implements the transaction interface for datastores who do
 // not have any sort of underlying transactional support.
 type SimpleTx struct {
-	ops    map[datastore.Key]op
+	ops    map[ds.Key]op
 	lock   sync.RWMutex
-	target datastore.Datastore
+	target ds.Datastore
 }
 
-func NewSimpleTx(ds datastore.Datastore) datastore.Txn {
+func NewSimpleTx(store ds.Datastore) dse.TxnExt {
 	return &SimpleTx{
-		ops:    make(map[datastore.Key]op),
-		target: ds,
+		ops:    make(map[ds.Key]op),
+		target: store,
 	}
 }
 
@@ -53,32 +68,38 @@ func (bt *SimpleTx) Query(q query.Query) (query.Results, error) {
 	return bt.target.Query(q)
 }
 
-func (bt *SimpleTx) Get(k datastore.Key) ([]byte, error) {
+func (bt *SimpleTx) QueryExtended(q dse.QueryExt) (query.Results, error) {
+	bt.lock.RLock()
+	defer bt.lock.RUnlock()
+	return bt.target.Query(q.Query)
+}
+
+func (bt *SimpleTx) Get(k ds.Key) ([]byte, error) {
 	bt.lock.RLock()
 	defer bt.lock.RUnlock()
 	return bt.target.Get(k)
 }
 
-func (bt *SimpleTx) Has(k datastore.Key) (bool, error) {
+func (bt *SimpleTx) Has(k ds.Key) (bool, error) {
 	bt.lock.RLock()
 	defer bt.lock.RUnlock()
 	return bt.target.Has(k)
 }
 
-func (bt *SimpleTx) GetSize(k datastore.Key) (int, error) {
+func (bt *SimpleTx) GetSize(k ds.Key) (int, error) {
 	bt.lock.RLock()
 	defer bt.lock.RUnlock()
 	return bt.target.GetSize(k)
 }
 
-func (bt *SimpleTx) Put(key datastore.Key, val []byte) error {
+func (bt *SimpleTx) Put(key ds.Key, val []byte) error {
 	bt.lock.RLock()
 	defer bt.lock.RUnlock()
 	bt.ops[key] = op{value: val}
 	return nil
 }
 
-func (bt *SimpleTx) Delete(key datastore.Key) error {
+func (bt *SimpleTx) Delete(key ds.Key) error {
 	bt.lock.RLock()
 	defer bt.lock.RUnlock()
 	bt.ops[key] = op{delete: true}
diff --git a/db/db.go b/db/db.go
index f25f0f3..c28e63a 100644
--- a/db/db.go
+++ b/db/db.go
@@ -14,18 +14,18 @@ import (
 
 	"github.com/alecthomas/jsonschema"
 	"github.com/dop251/goja"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	format "github.com/ipfs/go-ipld-format"
 	logging "github.com/ipfs/go-log"
 	ma "github.com/multiformats/go-multiaddr"
-	ds "github.com/textileio/go-datastore"
-	kt "github.com/textileio/go-datastore/keytransform"
-	"github.com/textileio/go-datastore/query"
 	threadcbor "github.com/textileio/go-threads/cbor"
 	"github.com/textileio/go-threads/core/app"
 	core "github.com/textileio/go-threads/core/db"
 	lstore "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/core/net"
 	"github.com/textileio/go-threads/core/thread"
+	kt "github.com/textileio/go-threads/db/keytransform"
 	"github.com/textileio/go-threads/util"
 )
 
@@ -78,7 +78,7 @@ type DB struct {
 	name      string
 	connector *app.Connector
 
-	datastore  ds.TxnDatastore
+	datastore  kt.TxnDatastoreExtended
 	dispatcher *dispatcher
 	eventcodec core.EventCodec
 
@@ -93,11 +93,24 @@ type DB struct {
 
 // NewDB creates a new DB, which will *own* ds and dispatcher for internal use.
 // Saying it differently, ds and dispatcher shouldn't be used externally.
-func NewDB(ctx context.Context, network app.Net, id thread.ID, opts ...NewOption) (*DB, error) {
+func NewDB(
+	ctx context.Context,
+	store kt.TxnDatastoreExtended,
+	network app.Net,
+	id thread.ID,
+	opts ...NewOption,
+) (*DB, error) {
 	args := &NewOptions{}
 	for _, opt := range opts {
 		opt(args)
 	}
+	if args.Debug {
+		if err := util.SetLogLevels(map[string]logging.LogLevel{
+			"db": logging.LevelDebug,
+		}); err != nil {
+			return nil, err
+		}
+	}
 
 	if args.Key.Defined() && !args.Key.CanRead() {
 		return nil, ErrThreadReadKeyRequired
@@ -111,7 +124,7 @@ func NewDB(ctx context.Context, network app.Net, id thread.ID, opts ...NewOption
 	); err != nil && !errors.Is(err, lstore.ErrThreadExists) && !errors.Is(err, lstore.ErrLogExists) {
 		return nil, err
 	}
-	return newDB(network, id, args)
+	return newDB(store, network, id, args)
 }
 
 // NewDBFromAddr creates a new DB from a thread hosted by another peer at address,
@@ -119,6 +132,7 @@ func NewDB(ctx context.Context, network app.Net, id thread.ID, opts ...NewOption
 // Saying it differently, ds and dispatcher shouldn't be used externally.
 func NewDBFromAddr(
 	ctx context.Context,
+	store kt.TxnDatastoreExtended,
 	network app.Net,
 	addr ma.Multiaddr,
 	key thread.Key,
@@ -128,11 +142,18 @@ func NewDBFromAddr(
 	for _, opt := range opts {
 		opt(args)
 	}
+	if args.Debug {
+		if err := util.SetLogLevels(map[string]logging.LogLevel{
+			"db": logging.LevelDebug,
+		}); err != nil {
+			return nil, err
+		}
+	}
 
 	if key.Defined() && !key.CanRead() {
 		return nil, ErrThreadReadKeyRequired
 	}
-	ti, err := network.AddThread(
+	info, err := network.AddThread(
 		ctx,
 		addr,
 		net.WithThreadKey(key),
@@ -142,21 +163,21 @@ func NewDBFromAddr(
 	if err != nil {
 		return nil, err
 	}
-	d, err := newDB(network, ti.ID, args)
+	d, err := newDB(store, network, info.ID, args)
 	if err != nil {
 		return nil, err
 	}
 
 	if args.Block {
-		if err = network.PullThread(ctx, ti.ID, net.WithThreadToken(args.Token)); err != nil {
+		if err = network.PullThread(ctx, info.ID, net.WithThreadToken(args.Token)); err != nil {
 			return nil, err
 		}
 	} else {
 		go func() {
 			ctx, cancel := context.WithTimeout(context.Background(), pullThreadBackgroundTimeout)
 			defer cancel()
-			if err := network.PullThread(ctx, ti.ID, net.WithThreadToken(args.Token)); err != nil {
-				log.Errorf("error pulling thread %s", ti.ID)
+			if err := network.PullThread(ctx, info.ID, net.WithThreadToken(args.Token)); err != nil {
+				log.Errorf("error pulling thread %s", info.ID)
 			}
 		}()
 	}
@@ -164,30 +185,14 @@ func NewDBFromAddr(
 }
 
 // newDB is used directly by a db manager to create new dbs with the same config.
-func newDB(n app.Net, id thread.ID, opts *NewOptions) (*DB, error) {
-	if opts.Datastore == nil {
-		datastore, err := newDefaultDatastore(opts.RepoPath, opts.LowMem)
-		if err != nil {
-			return nil, err
-		}
-		opts.Datastore = datastore
-	}
+func newDB(s kt.TxnDatastoreExtended, n app.Net, id thread.ID, opts *NewOptions) (*DB, error) {
 	if opts.EventCodec == nil {
 		opts.EventCodec = newDefaultEventCodec()
 	}
-	if !managedDatastore(opts.Datastore) {
-		if opts.Debug {
-			if err := util.SetLogLevels(map[string]logging.LogLevel{
-				"db": logging.LevelDebug,
-			}); err != nil {
-				return nil, err
-			}
-		}
-	}
 
 	d := &DB{
-		datastore:           opts.Datastore,
-		dispatcher:          newDispatcher(opts.Datastore),
+		datastore:           s,
+		dispatcher:          newDispatcher(s),
 		eventcodec:          opts.EventCodec,
 		collections:         make(map[string]*Collection),
 		localEventsBus:      app.NewLocalEventsBus(),
@@ -224,13 +229,6 @@ func newDB(n app.Net, id thread.ID, opts *NewOptions) (*DB, error) {
 	return d, nil
 }
 
-// managedDatastore returns whether or not the datastore is
-// being wrapped by an external datastore.
-func managedDatastore(ds ds.Datastore) bool {
-	_, ok := ds.(kt.KeyTransform)
-	return ok
-}
-
 // saveName saves the db name.
 func (d *DB) saveName(prevName string) error {
 	if d.name == prevName {
@@ -540,13 +538,7 @@ func (d *DB) Close() error {
 		return nil
 	}
 	d.closed = true
-
 	d.localEventsBus.Discard()
-	if !managedDatastore(d.datastore) {
-		if err := d.datastore.Close(); err != nil {
-			return err
-		}
-	}
 	d.stateChangedNotifee.close()
 	return nil
 }
diff --git a/db/dispatcher.go b/db/dispatcher.go
index bda383f..6a6f96e 100644
--- a/db/dispatcher.go
+++ b/db/dispatcher.go
@@ -8,9 +8,10 @@ import (
 	"strconv"
 	"sync"
 
-	datastore "github.com/textileio/go-datastore"
-	"github.com/textileio/go-datastore/query"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	core "github.com/textileio/go-threads/core/db"
+	kt "github.com/textileio/go-threads/db/keytransform"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -29,21 +30,21 @@ type Reducer interface {
 // Every event is dispatched to every registered reducer. When a given reducer is registered, it returns a `token`,
 // which can be used to deregister the reducer later.
 type dispatcher struct {
-	store    datastore.TxnDatastore
+	store    kt.TxnDatastoreExtended
 	reducers []Reducer
 	lock     sync.RWMutex
 	lastID   int
 }
 
 // NewDispatcher creates a new EventDispatcher.
-func newDispatcher(store datastore.TxnDatastore) *dispatcher {
+func newDispatcher(store kt.TxnDatastoreExtended) *dispatcher {
 	return &dispatcher{
 		store: store,
 	}
 }
 
 // Store returns the internal event store.
-func (d *dispatcher) Store() datastore.TxnDatastore {
+func (d *dispatcher) Store() kt.TxnDatastoreExtended {
 	return d.store
 }
 
@@ -107,7 +108,7 @@ func (d *dispatcher) Dispatch(events []core.Event) error {
 }
 
 // Query searches the internal event store and returns a query result.
-// This is a syncronouse version of github.com/textileio/go-datastore's Query method.
+// This is a synchronous version of github.com/ipfs/go-datastore's Query method.
 func (d *dispatcher) Query(query query.Query) ([]query.Entry, error) {
 	result, err := d.store.Query(query)
 	if err != nil {
@@ -118,7 +119,7 @@ func (d *dispatcher) Query(query query.Query) ([]query.Entry, error) {
 
 // Key format: <timestamp>/<instance-id>/<type>
 // @todo: This is up for debate, its a 'fake' Event struct right now anyway
-func getKey(event core.Event) (key datastore.Key, err error) {
+func getKey(event core.Event) (key ds.Key, err error) {
 	buf := bytes.NewBuffer(event.Time())
 	var unix int64
 	if err = binary.Read(buf, binary.BigEndian, &unix); err != nil {
diff --git a/db/index.go b/db/index.go
index 3dd8a32..6016ede 100644
--- a/db/index.go
+++ b/db/index.go
@@ -11,9 +11,11 @@ import (
 	"fmt"
 	"sort"
 
+	dse "github.com/textileio/go-datastore-extensions"
+
 	"github.com/alecthomas/jsonschema"
-	ds "github.com/textileio/go-datastore"
-	"github.com/textileio/go-datastore/query"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	"github.com/tidwall/gjson"
 	"github.com/tidwall/sjson"
 )
@@ -274,7 +276,7 @@ type iterator struct {
 	iter     query.Results
 }
 
-func newIterator(txn ds.Txn, baseKey ds.Key, q *Query) *iterator {
+func newIterator(txn dse.TxnExt, baseKey ds.Key, q *Query) *iterator {
 	i := &iterator{
 		txn:   txn,
 		query: q,
@@ -286,10 +288,12 @@ func newIterator(txn ds.Txn, baseKey ds.Key, q *Query) *iterator {
 		prefix = indexPrefix.Child(baseKey).ChildString(q.Index)
 	}
 
-	dsq := query.Query{
-		Prefix: prefix.String(),
-		Limit:  q.Limit,
-		Offset: q.Skip,
+	dsq := dse.QueryExt{
+		Query: query.Query{
+			Prefix: prefix.String(),
+			Limit:  q.Limit,
+			Offset: q.Skip,
+		},
 	}
 	if q.Sort.FieldPath == idFieldName {
 		if q.Sort.Desc {
@@ -301,7 +305,7 @@ func newIterator(txn ds.Txn, baseKey ds.Key, q *Query) *iterator {
 	if q.Seek != "" {
 		dsq.SeekPrefix = prefix.Child(ds.NewKey(string(q.Seek))).String()
 	}
-	i.iter, i.err = txn.Query(dsq)
+	i.iter, i.err = txn.QueryExtended(dsq)
 
 	// Key field or index not specified, pass thru to base 'iterator'
 	if q.Index == "" {
diff --git a/db/keytransform.go b/db/keytransform/keytransform.go
similarity index 72%
rename from db/keytransform.go
rename to db/keytransform/keytransform.go
index a1e5e27..eb3de64 100644
--- a/db/keytransform.go
+++ b/db/keytransform/keytransform.go
@@ -1,34 +1,53 @@
-package db
+package keytransform
 
 import (
-	ds "github.com/textileio/go-datastore"
-	kt "github.com/textileio/go-datastore/keytransform"
-	dsq "github.com/textileio/go-datastore/query"
+	ds "github.com/ipfs/go-datastore"
+	kt "github.com/ipfs/go-datastore/keytransform"
+	dsq "github.com/ipfs/go-datastore/query"
+	dse "github.com/textileio/go-datastore-extensions"
 )
 
-func wrapTxnDatastore(child ds.TxnDatastore, t kt.KeyTransform) *Datastore {
-	nds := &Datastore{
+// TxnDatastoreExtended adds QueryExtensions to TxnDatastore.
+type TxnDatastoreExtended interface {
+	ds.TxnDatastore
+	dse.DatastoreExtensions
+}
+
+// WrapTxnDatastore wraps a datastore with a key transform.
+func WrapTxnDatastore(child TxnDatastoreExtended, t kt.KeyTransform) *Datastore {
+	return &Datastore{
 		child:        child,
 		Datastore:    kt.Wrap(child, t),
 		KeyTransform: t,
 	}
-	return nds
 }
 
 // Datastore keeps a KeyTransform function
 type Datastore struct {
-	child ds.TxnDatastore
-	kt.KeyTransform
+	child TxnDatastoreExtended
 	ds.Datastore
+	kt.KeyTransform
 }
 
+var _ dse.DatastoreExtensions = (*Datastore)(nil)
+
 type txn struct {
-	ds.Txn
-	ds *Datastore
+	Txn dse.TxnExt
+	ds  *Datastore
 }
 
+var _ dse.TxnExt = (*txn)(nil)
+
 func (d *Datastore) NewTransaction(readOnly bool) (ds.Txn, error) {
-	t, err := d.child.NewTransaction(readOnly)
+	return d.newTransaction(readOnly)
+}
+
+func (d *Datastore) NewTransactionExtended(readOnly bool) (dse.TxnExt, error) {
+	return d.newTransaction(readOnly)
+}
+
+func (d *Datastore) newTransaction(readOnly bool) (dse.TxnExt, error) {
+	t, err := d.child.NewTransactionExtended(readOnly)
 	if err != nil {
 		return nil, err
 	}
@@ -36,6 +55,10 @@ func (d *Datastore) NewTransaction(readOnly bool) (ds.Txn, error) {
 	return &txn{Txn: t, ds: d}, nil
 }
 
+func (d *Datastore) QueryExtended(q dse.QueryExt) (dsq.Results, error) {
+	return d.child.QueryExtended(q)
+}
+
 func (t *txn) Commit() error {
 	return t.Txn.Commit()
 }
@@ -73,34 +96,29 @@ func (t *txn) GetSize(key ds.Key) (size int, err error) {
 
 // Query implements Query, inverting keys on the way back out.
 func (t *txn) Query(q dsq.Query) (dsq.Results, error) {
-	nq, cq := t.prepareQuery(q)
+	nq, cq := t.prepareQuery(dse.QueryExt{Query: q})
 
-	cqr, err := t.Txn.Query(cq)
+	qr, err := t.Txn.Query(cq.Query)
 	if err != nil {
 		return nil, err
 	}
+	return dsq.NaiveQueryApply(nq.Query, dsq.ResultsFromIterator(q, t.getIterator(qr))), nil
+}
 
-	qr := dsq.ResultsFromIterator(q, dsq.Iterator{
-		Next: func() (dsq.Result, bool) {
-			r, ok := cqr.NextSync()
-			if !ok {
-				return r, false
-			}
-			if r.Error == nil {
-				r.Entry.Key = t.ds.InvertKey(ds.RawKey(r.Entry.Key)).String()
-			}
-			return r, true
-		},
-		Close: func() error {
-			return cqr.Close()
-		},
-	})
-	return dsq.NaiveQueryApply(nq, qr), nil
+// QueryExtended implements QueryExtended, inverting keys on the way back out.
+func (t *txn) QueryExtended(q dse.QueryExt) (dsq.Results, error) {
+	nq, cq := t.prepareQuery(q)
+
+	qr, err := t.Txn.QueryExtended(cq)
+	if err != nil {
+		return nil, err
+	}
+	return dsq.NaiveQueryApply(nq.Query, dsq.ResultsFromIterator(q.Query, t.getIterator(qr))), nil
 }
 
 // Split the query into a child query and a naive query. That way, we can make
 // the child datastore do as much work as possible.
-func (t *txn) prepareQuery(q dsq.Query) (naive, child dsq.Query) {
+func (t *txn) prepareQuery(q dse.QueryExt) (naive, child dse.QueryExt) {
 
 	// First, put everything in the child query. Then, start taking things
 	// out.
@@ -199,4 +217,20 @@ orders:
 	return
 }
 
-var _ ds.TxnDatastore = (*Datastore)(nil)
+func (t *txn) getIterator(qr dsq.Results) dsq.Iterator {
+	return dsq.Iterator{
+		Next: func() (dsq.Result, bool) {
+			r, ok := qr.NextSync()
+			if !ok {
+				return r, false
+			}
+			if r.Error == nil {
+				r.Entry.Key = t.ds.InvertKey(ds.RawKey(r.Entry.Key)).String()
+			}
+			return r, true
+		},
+		Close: func() error {
+			return qr.Close()
+		},
+	}
+}
diff --git a/db/manager.go b/db/manager.go
index 0a3185a..936ba40 100644
--- a/db/manager.go
+++ b/db/manager.go
@@ -6,14 +6,15 @@ import (
 	"io"
 	"strings"
 
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/keytransform"
+	"github.com/ipfs/go-datastore/query"
 	logging "github.com/ipfs/go-log"
 	ma "github.com/multiformats/go-multiaddr"
-	ds "github.com/textileio/go-datastore"
-	kt "github.com/textileio/go-datastore/keytransform"
-	"github.com/textileio/go-datastore/query"
 	"github.com/textileio/go-threads/core/app"
 	"github.com/textileio/go-threads/core/net"
 	"github.com/textileio/go-threads/core/thread"
+	kt "github.com/textileio/go-threads/db/keytransform"
 	"github.com/textileio/go-threads/util"
 )
 
@@ -31,25 +32,19 @@ type Manager struct {
 
 	opts *NewOptions
 
+	store   kt.TxnDatastoreExtended
 	network app.Net
 	dbs     map[thread.ID]*DB
 }
 
 // NewManager hydrates and starts dbs from prefixes.
-func NewManager(network app.Net, opts ...NewOption) (*Manager, error) {
-	options := &NewOptions{}
+func NewManager(store kt.TxnDatastoreExtended, network app.Net, opts ...NewOption) (*Manager, error) {
+	args := &NewOptions{}
 	for _, opt := range opts {
-		opt(options)
+		opt(args)
 	}
 
-	if options.Datastore == nil {
-		datastore, err := newDefaultDatastore(options.RepoPath, options.LowMem)
-		if err != nil {
-			return nil, err
-		}
-		options.Datastore = datastore
-	}
-	if options.Debug {
+	if args.Debug {
 		if err := util.SetLogLevels(map[string]logging.LogLevel{
 			"db": logging.LevelDebug,
 		}); err != nil {
@@ -58,12 +53,13 @@ func NewManager(network app.Net, opts ...NewOption) (*Manager, error) {
 	}
 
 	m := &Manager{
-		opts:    options,
+		store:   store,
 		network: network,
+		opts:    args,
 		dbs:     make(map[thread.ID]*DB),
 	}
 
-	results, err := m.opts.Datastore.Query(query.Query{
+	results, err := store.Query(query.Query{
 		Prefix:   dsManagerBaseKey.String(),
 		KeysOnly: true,
 	})
@@ -87,17 +83,17 @@ func NewManager(network app.Net, opts ...NewOption) (*Manager, error) {
 		if _, ok := invalids[id]; ok {
 			continue
 		}
-		opts, err := getDBOptions(id, m.opts, "")
+		s, opts, err := wrapDB(store, id, m.opts, "")
 		if err != nil {
 			return nil, err
 		}
-		s, err := newDB(m.network, id, opts)
+		d, err := newDB(s, m.network, id, opts)
 		if err != nil {
 			log.Errorf("unable to reload db %s: %s (marked for deletion)", id, err)
 			invalids[id] = struct{}{}
 			continue
 		}
-		m.dbs[id] = s
+		m.dbs[id] = d
 	}
 
 	// Cleanup invalids
@@ -140,11 +136,11 @@ func (m *Manager) NewDB(ctx context.Context, id thread.ID, opts ...NewManagedOpt
 		return nil, err
 	}
 
-	dbOpts, err := getDBOptions(id, m.opts, args.Name, args.Collections...)
+	store, dbOpts, err := wrapDB(m.store, id, m.opts, args.Name, args.Collections...)
 	if err != nil {
 		return nil, err
 	}
-	db, err := newDB(m.network, id, dbOpts)
+	db, err := newDB(store, m.network, id, dbOpts)
 	if err != nil {
 		return nil, err
 	}
@@ -189,11 +185,11 @@ func (m *Manager) NewDBFromAddr(
 		return nil, err
 	}
 
-	dbOpts, err := getDBOptions(id, m.opts, args.Name, args.Collections...)
+	store, dbOpts, err := wrapDB(m.store, id, m.opts, args.Name, args.Collections...)
 	if err != nil {
 		return nil, err
 	}
-	db, err := newDB(m.network, id, dbOpts)
+	db, err := newDB(store, m.network, id, dbOpts)
 	if err != nil {
 		return nil, err
 	}
@@ -289,13 +285,13 @@ func (m *Manager) DeleteDB(ctx context.Context, id thread.ID, opts ...ManagedOpt
 func (m *Manager) deleteThreadNamespace(id thread.ID) error {
 	pre := dsManagerBaseKey.ChildString(id.String())
 	q := query.Query{Prefix: pre.String(), KeysOnly: true}
-	results, err := m.opts.Datastore.Query(q)
+	results, err := m.store.Query(q)
 	if err != nil {
 		return err
 	}
 	defer results.Close()
 	for result := range results.Next() {
-		if err := m.opts.Datastore.Delete(ds.NewKey(result.Key)); err != nil {
+		if err := m.store.Delete(ds.NewKey(result.Key)); err != nil {
 			return err
 		}
 	}
@@ -314,25 +310,30 @@ func (m *Manager) Close() error {
 			log.Error("error when closing manager datastore: %v", err)
 		}
 	}
-	return m.opts.Datastore.Close()
+	return nil
 }
 
-// getDBOptions copies the manager's base config,
+// wrapDB copies the manager's base config,
 // wraps the datastore with an id prefix,
 // and merges specified collection configs with those from base
-func getDBOptions(id thread.ID, base *NewOptions, name string, collections ...CollectionConfig) (*NewOptions, error) {
+func wrapDB(
+	store kt.TxnDatastoreExtended,
+	id thread.ID,
+	base *NewOptions,
+	name string,
+	collections ...CollectionConfig,
+) (kt.TxnDatastoreExtended, *NewOptions, error) {
 	if err := id.Validate(); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	return &NewOptions{
-		Name:     name,
-		RepoPath: base.RepoPath,
-		Datastore: wrapTxnDatastore(base.Datastore, kt.PrefixTransform{
-			Prefix: dsManagerBaseKey.ChildString(id.String()),
-		}),
+	store = kt.WrapTxnDatastore(store, keytransform.PrefixTransform{
+		Prefix: dsManagerBaseKey.ChildString(id.String()),
+	})
+	opts := &NewOptions{
+		Name:        name,
 		Collections: append(base.Collections, collections...),
 		EventCodec:  base.EventCodec,
-		LowMem:      base.LowMem,
 		Debug:       base.Debug,
-	}, nil
+	}
+	return store, opts, nil
 }
diff --git a/db/options.go b/db/options.go
index d584e45..b9103f4 100644
--- a/db/options.go
+++ b/db/options.go
@@ -1,77 +1,31 @@
 package db
 
 import (
-	"os"
-	"path/filepath"
-
-	"github.com/dgraph-io/badger/options"
 	"github.com/libp2p/go-libp2p-core/crypto"
-	ds "github.com/textileio/go-datastore"
-	badger "github.com/textileio/go-ds-badger"
 	core "github.com/textileio/go-threads/core/db"
 	"github.com/textileio/go-threads/core/thread"
 	"github.com/textileio/go-threads/jsonpatcher"
 )
 
-const (
-	defaultDatastorePath = "eventstore"
-)
-
 func newDefaultEventCodec() core.EventCodec {
 	return jsonpatcher.New()
 }
 
-func newDefaultDatastore(repoPath string, lowMem bool) (ds.TxnDatastore, error) {
-	path := filepath.Join(repoPath, defaultDatastorePath)
-	if err := os.MkdirAll(path, os.ModePerm); err != nil {
-		return nil, err
-	}
-	opts := badger.DefaultOptions
-	if lowMem {
-		opts.TableLoadingMode = options.FileIO
-	}
-	return badger.NewDatastore(path, &opts)
-}
-
 // NewOptions defines options for creating a new db.
 type NewOptions struct {
 	Name        string
-	RepoPath    string
 	Key         thread.Key
 	LogKey      crypto.Key
-	Datastore   ds.TxnDatastore
 	Collections []CollectionConfig
 	Block       bool
 	EventCodec  core.EventCodec
 	Token       thread.Token
-	LowMem      bool
 	Debug       bool
 }
 
 // NewOption specifies a new db option.
 type NewOption func(*NewOptions)
 
-// WithNewName sets the db name.
-func WithNewName(name string) NewOption {
-	return func(o *NewOptions) {
-		o.Name = name
-	}
-}
-
-// WithNewRepoPath sets the repo path.
-func WithNewRepoPath(path string) NewOption {
-	return func(o *NewOptions) {
-		o.RepoPath = path
-	}
-}
-
-// WithNewToken provides authorization for interacting with a db.
-func WithNewToken(t thread.Token) NewOption {
-	return func(o *NewOptions) {
-		o.Token = t
-	}
-}
-
 // WithNewKey provides control over thread keys to use with a db.
 func WithNewKey(key thread.Key) NewOption {
 	return func(o *NewOptions) {
@@ -89,6 +43,13 @@ func WithNewLogKey(key crypto.Key) NewOption {
 	}
 }
 
+// WithNewName sets the db name.
+func WithNewName(name string) NewOption {
+	return func(o *NewOptions) {
+		o.Name = name
+	}
+}
+
 // WithNewCollections is used to specify collections that
 // will be created.
 func WithNewCollections(cs ...CollectionConfig) NewOption {
@@ -115,10 +76,10 @@ func WithNewEventCodec(ec core.EventCodec) NewOption {
 	}
 }
 
-// WithNewLowMem specifies whether or not to use low memory settings.
-func WithNewLowMem(low bool) NewOption {
+// WithNewToken provides authorization for interacting with a db.
+func WithNewToken(t thread.Token) NewOption {
 	return func(o *NewOptions) {
-		o.LowMem = low
+		o.Token = t
 	}
 }
 
diff --git a/db/query.go b/db/query.go
index d5ce5a8..6b06041 100644
--- a/db/query.go
+++ b/db/query.go
@@ -297,7 +297,7 @@ func (t *Txn) Find(q *Query) ([][]byte, error) {
 	if err := q.Validate(); err != nil {
 		return nil, fmt.Errorf("invalid query: %s", err)
 	}
-	txn, err := t.collection.db.datastore.NewTransaction(true)
+	txn, err := t.collection.db.datastore.NewTransactionExtended(true)
 	if err != nil {
 		return nil, fmt.Errorf("error building internal query: %v", err)
 	}
diff --git a/docker-compose-dev.yml b/docker-compose-dev.yml
index a6b0823..a6915e5 100644
--- a/docker-compose-dev.yml
+++ b/docker-compose-dev.yml
@@ -1,19 +1,37 @@
 version: "3"
 services:
-  threads:
+  threaddb:
     build:
       context: .
     environment:
+      - THRDS_REPO
       - THRDS_HOSTADDR
-      - THRDS_APIADDR
-      - THRDS_APIPROXYADDR
+      - THRDS_APIADDR=/ip4/0.0.0.0/tcp/6006
+      - THRDS_APIPROXYADDR=/ip4/0.0.0.0/tcp/6007
       - THRDS_CONNLOWWATER
       - THRDS_CONNHIGHWATER
       - THRDS_CONNGRACEPERIOD
       - THRDS_KEEPALIVEINTERVAL
       - THRDS_ENABLENETPUBSUB
-      - THRDS_DEBUG
+      - THRDS_MONGOURI=mongodb://mongo:27017
+      - THRDS_MONGODATABASE=threaddb
+      - THRDS_DEBUG=true
     ports:
       - "4006:4006"
       - "127.0.0.1:6006:6006"
       - "127.0.0.1:6007:6007"
+    depends_on:
+      - mongo
+  mongo:
+    image: mongo:latest
+    expose:
+      - "27017"
+    ports:
+      - "127.0.0.1:27017:27017"
+    command:
+      - /bin/bash
+      - -c
+      - |
+        /usr/bin/mongod --fork --logpath /var/log/mongod.log --bind_ip_all --replSet rs0
+        mongo --eval 'rs.initiate({_id: "rs0", version: 1, members: [{ _id: 0, host: "mongo:27017" }]})'
+        tail -f /var/log/mongod.log
diff --git a/docker-compose.yml b/docker-compose.yml
index 8f8afa4..ffdbeee 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -4,18 +4,39 @@ services:
     image: textile/go-threads:latest
     restart: always
     volumes:
-      - "${REPO_PATH}/threads:/data/threads"
+      - "${THRDS_REPO}/threaddb:/data/threaddb"
     environment:
+      - THRDS_REPO
       - THRDS_HOSTADDR
-      - THRDS_APIADDR
-      - THRDS_APIPROXYADDR
+      - THRDS_APIADDR=/ip4/0.0.0.0/tcp/6006
+      - THRDS_APIPROXYADDR=/ip4/0.0.0.0/tcp/6007
       - THRDS_CONNLOWWATER
       - THRDS_CONNHIGHWATER
       - THRDS_CONNGRACEPERIOD
       - THRDS_KEEPALIVEINTERVAL
       - THRDS_ENABLENETPUBSUB
+      - THRDS_MONGOURI=mongodb://mongo:27017
+      - THRDS_MONGODATABASE=threaddb
       - THRDS_DEBUG
     ports:
       - "4006:4006"
       - "127.0.0.1:6006:6006"
       - "127.0.0.1:6007:6007"
+    depends_on:
+      - mongo
+  mongo:
+    image: mongo:latest
+    restart: always
+    volumes:
+      - "${THRDS_REPO}/mongodb:/data/mongodb"
+    expose:
+      - "27017"
+    ports:
+      - "127.0.0.1:27017:27017"
+    command:
+      - /bin/bash
+      - -c
+      - |
+        /usr/bin/mongod --fork --logpath /var/log/mongod.log --bind_ip_all --replSet rs0
+        mongo --eval 'rs.initiate({_id: "rs0", version: 1, members: [{ _id: 0, host: "mongo:27017" }]})'
+        tail -f /var/log/mongod.log
diff --git a/go.mod b/go.mod
index c26d221..c1b1552 100644
--- a/go.mod
+++ b/go.mod
@@ -1,14 +1,13 @@
 module github.com/textileio/go-threads
 
-go 1.14
+go 1.15
 
 require (
-	github.com/AndreasBriese/bbloom v0.0.0-20190823232136-616930265c33 // indirect
 	// agl/ed25519 only used in tests for backward compatibility, *do not* use in production code.
 	github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412
 	github.com/alecthomas/jsonschema v0.0.0-20191017121752-4bb6e3fae4f2
 	github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect
-	github.com/dgraph-io/badger v1.6.1
+	github.com/dgraph-io/badger v1.6.2
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/dlclark/regexp2 v1.2.0 // indirect
 	github.com/dop251/goja v0.0.0-20200721192441-a695b0cdd498
@@ -27,8 +26,7 @@ require (
 	github.com/ipfs/go-block-format v0.0.2
 	github.com/ipfs/go-blockservice v0.1.3
 	github.com/ipfs/go-cid v0.0.7
-	github.com/ipfs/go-datastore v0.4.4
-	github.com/ipfs/go-ds-badger v0.2.4
+	github.com/ipfs/go-datastore v0.4.5
 	github.com/ipfs/go-ipfs-blockstore v1.0.1
 	github.com/ipfs/go-ipfs-exchange-offline v0.0.1
 	github.com/ipfs/go-ipfs-files v0.0.4 // indirect
@@ -41,13 +39,9 @@ require (
 	github.com/libp2p/go-libp2p v0.10.3
 	github.com/libp2p/go-libp2p-connmgr v0.2.4
 	github.com/libp2p/go-libp2p-core v0.6.1
-	github.com/libp2p/go-libp2p-crypto v0.1.0
 	github.com/libp2p/go-libp2p-gostream v0.2.0
-	github.com/libp2p/go-libp2p-kad-dht v0.8.3
-	github.com/libp2p/go-libp2p-peer v0.2.0
 	github.com/libp2p/go-libp2p-peerstore v0.2.6
 	github.com/libp2p/go-libp2p-pubsub v0.2.4
-	github.com/libp2p/go-libp2p-swarm v0.2.8
 	github.com/multiformats/go-multiaddr v0.2.2
 	github.com/multiformats/go-multibase v0.0.3
 	github.com/multiformats/go-multihash v0.0.14
@@ -59,8 +53,9 @@ require (
 	github.com/polydawn/refmt v0.0.0-20190807091052-3d65705ee9f1 // indirect
 	github.com/rs/cors v1.7.0 // indirect
 	github.com/smartystreets/goconvey v0.0.0-20190710185942-9d28bd7c0945 // indirect
-	github.com/textileio/go-datastore v0.4.5-0.20200819232101-baa577bf9422
-	github.com/textileio/go-ds-badger v0.2.5-0.20200819232634-de89720b5d6a
+	github.com/textileio/go-datastore-extensions v1.0.1
+	github.com/textileio/go-ds-badger v0.2.7-0.20201204225019-4ee78c4a40e2
+	github.com/textileio/go-ds-mongo v0.1.4-0.20201205032204-63f494d9c3f3
 	github.com/tidwall/gjson v1.3.5
 	github.com/tidwall/sjson v1.0.4
 	github.com/whyrusleeping/base32 v0.0.0-20170828182744-c30ac30633cc
@@ -70,8 +65,6 @@ require (
 	golang.org/x/crypto v0.0.0-20200423211502-4bdfaf469ed5
 	golang.org/x/exp v0.0.0-20200331195152-e8c3332aa8e5
 	golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a
-	golang.org/x/text v0.3.3 // indirect
 	google.golang.org/genproto v0.0.0-20200428115010-c45acf45369a // indirect
 	google.golang.org/grpc v1.31.0
-	google.golang.org/protobuf v1.23.0
 )
diff --git a/go.sum b/go.sum
index 3d0810f..d34ea33 100644
--- a/go.sum
+++ b/go.sum
@@ -12,8 +12,8 @@ git.apache.org/thrift.git v0.0.0-20180902110319-2566ecd5d999/go.mod h1:fPE2ZNJGy
 github.com/AndreasBriese/bbloom v0.0.0-20180913140656-343706a395b7/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9 h1:HD8gA2tkByhMAwYaFAX9w2l7vxvBQ5NMoxDrkhqhtn4=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
-github.com/AndreasBriese/bbloom v0.0.0-20190823232136-616930265c33 h1:2/E2IVdZoHh/aCBq4Gchy2MGWkTmbReP46/Wnt9qhKs=
-github.com/AndreasBriese/bbloom v0.0.0-20190823232136-616930265c33/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
+github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96 h1:cTp8I5+VIoKjsnZuH8vjyaysT/ses3EvZeaV/1UkF2M=
+github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
@@ -31,6 +31,8 @@ github.com/alecthomas/jsonschema v0.0.0-20191017121752-4bb6e3fae4f2/go.mod h1:Ju
 github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/awalterschulze/gographviz v0.0.0-20190522210029-fa59802746ab/go.mod h1:GEV5wmg4YquNw7v1kkyoX9etIk8yVmXj+AkDHuuETHs=
+github.com/aws/aws-sdk-go v1.29.15 h1:0ms/213murpsujhsnxnNKNeVouW60aJqSd992Ks3mxs=
+github.com/aws/aws-sdk-go v1.29.15/go.mod h1:1KvfttTE3SPKMpo8g2c6jL3ZKfXtFvKscTgahTma5Xg=
 github.com/benbjohnson/clock v1.0.2 h1:Z0CN0Yb4ig9sGPXkvAQcGJfnrrMQ5QYLCMPRi9iD7YE=
 github.com/benbjohnson/clock v1.0.2/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
@@ -85,6 +87,8 @@ github.com/dgraph-io/badger v1.6.0 h1:DshxFxZWXUcO0xX476VJC07Xsr6ZCBVRHKZ93Oh7Ev
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgraph-io/badger v1.6.1 h1:w9pSFNSdq/JPM1N12Fz/F/bzo993Is1W+Q7HjPzi7yg=
 github.com/dgraph-io/badger v1.6.1/go.mod h1:FRmFw3uxvcpa8zG3Rxs0th+hCLIuaQg8HlNV5bjgnuU=
+github.com/dgraph-io/badger v1.6.2 h1:mNw0qs90GVgGGWylh0umH5iag1j6n/PeJtNvL6KY/x8=
+github.com/dgraph-io/badger v1.6.2/go.mod h1:JW2yswe3V058sS0kZ2h/AXeDSqFjxnZcRrVH//y2UQE=
 github.com/dgraph-io/ristretto v0.0.2 h1:a5WaUrDa0qm0YrAAS1tUykT5El3kt62KNZZeMxQn3po=
 github.com/dgraph-io/ristretto v0.0.2/go.mod h1:KPxhHT9ZxKefz+PCeOGsrHpl1qZ7i70dGTu2u+Ahh6E=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
@@ -121,7 +125,33 @@ github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-sourcemap/sourcemap v2.1.3+incompatible h1:W1iEw64niKVGogNgBN3ePyLFfuisuzeidWPMPWmECqU=
 github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
+github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
+github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/gobuffalo/attrs v0.0.0-20190224210810-a9411de4debd/go.mod h1:4duuawTqi2wkkpB4ePgWMaai6/Kc6WEz83bhFwpHzj0=
+github.com/gobuffalo/depgen v0.0.0-20190329151759-d478694a28d3/go.mod h1:3STtPUQYuzV0gBVOY3vy6CfMm/ljR4pABfrTeHNLHUY=
+github.com/gobuffalo/depgen v0.1.0/go.mod h1:+ifsuy7fhi15RWncXQQKjWS9JPkdah5sZvtHc2RXGlg=
+github.com/gobuffalo/envy v1.6.15/go.mod h1:n7DRkBerg/aorDM8kbduw5dN3oXGswK5liaSCx4T5NI=
+github.com/gobuffalo/envy v1.7.0/go.mod h1:n7DRkBerg/aorDM8kbduw5dN3oXGswK5liaSCx4T5NI=
+github.com/gobuffalo/flect v0.1.0/go.mod h1:d2ehjJqGOH/Kjqcoz+F7jHTBbmDb38yXA598Hb50EGs=
+github.com/gobuffalo/flect v0.1.1/go.mod h1:8JCgGVbRjJhVgD6399mQr4fx5rRfGKVzFjbj6RE/9UI=
+github.com/gobuffalo/flect v0.1.3/go.mod h1:8JCgGVbRjJhVgD6399mQr4fx5rRfGKVzFjbj6RE/9UI=
+github.com/gobuffalo/genny v0.0.0-20190329151137-27723ad26ef9/go.mod h1:rWs4Z12d1Zbf19rlsn0nurr75KqhYp52EAGGxTbBhNk=
+github.com/gobuffalo/genny v0.0.0-20190403191548-3ca520ef0d9e/go.mod h1:80lIj3kVJWwOrXWWMRzzdhW3DsrdjILVil/SFKBzF28=
+github.com/gobuffalo/genny v0.1.0/go.mod h1:XidbUqzak3lHdS//TPu2OgiFB+51Ur5f7CSnXZ/JDvo=
+github.com/gobuffalo/genny v0.1.1/go.mod h1:5TExbEyY48pfunL4QSXxlDOmdsD44RRq4mVZ0Ex28Xk=
+github.com/gobuffalo/gitgen v0.0.0-20190315122116-cc086187d211/go.mod h1:vEHJk/E9DmhejeLeNt7UVvlSGv3ziL+djtTr3yyzcOw=
+github.com/gobuffalo/gogen v0.0.0-20190315121717-8f38393713f5/go.mod h1:V9QVDIxsgKNZs6L2IYiGR8datgMhB577vzTDqypH360=
+github.com/gobuffalo/gogen v0.1.0/go.mod h1:8NTelM5qd8RZ15VjQTFkAW6qOMx5wBbW4dSCS3BY8gg=
+github.com/gobuffalo/gogen v0.1.1/go.mod h1:y8iBtmHmGc4qa3urIyo1shvOD8JftTtfcKi+71xfDNE=
+github.com/gobuffalo/logger v0.0.0-20190315122211-86e12af44bc2/go.mod h1:QdxcLw541hSGtBnhUc4gaNIXRjiDppFGaDqzbrBd3v8=
+github.com/gobuffalo/mapi v1.0.1/go.mod h1:4VAGh89y6rVOvm5A8fKFxYG+wIW6LO1FMTG9hnKStFc=
+github.com/gobuffalo/mapi v1.0.2/go.mod h1:4VAGh89y6rVOvm5A8fKFxYG+wIW6LO1FMTG9hnKStFc=
+github.com/gobuffalo/packd v0.0.0-20190315124812-a385830c7fc0/go.mod h1:M2Juc+hhDXf/PnmBANFCqx4DM3wRbgDvnVWeG2RIxq4=
+github.com/gobuffalo/packd v0.1.0/go.mod h1:M2Juc+hhDXf/PnmBANFCqx4DM3wRbgDvnVWeG2RIxq4=
+github.com/gobuffalo/packr/v2 v2.0.9/go.mod h1:emmyGweYTm6Kdper+iywB6YK5YzuKchGtJQZ0Odn4pQ=
+github.com/gobuffalo/packr/v2 v2.2.0/go.mod h1:CaAwI0GPIAv+5wKLtv8Afwl+Cm78K/I/VCm/3ptBN+0=
+github.com/gobuffalo/syncx v0.0.0-20190224160051-33c29581e754/go.mod h1:HhnNqWY95UYwwW3uSASeV7vtgYkT2t16hJgV3AEPUpw=
 github.com/gogo/googleapis v0.0.0-20180223154316-0cd9801be74a h1:dR8+Q0uO5S2ZBcs2IH6VBKYwSxPo2vYCYq0ot0mu7xA=
 github.com/gogo/googleapis v0.0.0-20180223154316-0cd9801be74a/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
 github.com/gogo/googleapis v1.3.1 h1:CzMaKrvF6Qa7XtRii064vKBQiyvmY8H8vG1xa1/W1JA=
@@ -160,6 +190,8 @@ github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
@@ -259,6 +291,8 @@ github.com/ipfs/go-datastore v0.4.1/go.mod h1:SX/xMIKoCszPqp+z9JhPYCmoOoXTvaa13X
 github.com/ipfs/go-datastore v0.4.2/go.mod h1:SX/xMIKoCszPqp+z9JhPYCmoOoXTvaa13XEbGtsFUhA=
 github.com/ipfs/go-datastore v0.4.4 h1:rjvQ9+muFaJ+QZ7dN5B1MSDNQ0JVZKkkES/rMZmA8X8=
 github.com/ipfs/go-datastore v0.4.4/go.mod h1:SX/xMIKoCszPqp+z9JhPYCmoOoXTvaa13XEbGtsFUhA=
+github.com/ipfs/go-datastore v0.4.5 h1:cwOUcGMLdLPWgu3SlrCckCMznaGADbPqE0r8h768/Dg=
+github.com/ipfs/go-datastore v0.4.5/go.mod h1:eXTcaaiN6uOlVCLS9GjJUJtlvJfM3xk23w3fyfrmmJs=
 github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk=
 github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps=
 github.com/ipfs/go-ds-badger v0.0.2/go.mod h1:Y3QpeSFWQf6MopLTiZD+VT6IC1yZqaGmjvRcKeSGij8=
@@ -390,6 +424,9 @@ github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZl
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
 github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
+github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
+github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
+github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=
 github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
@@ -398,12 +435,18 @@ github.com/jtolds/gls v4.2.1+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVY
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/kami-zh/go-capturer v0.0.0-20171211120116-e492ea43421d/go.mod h1:P2viExyCEfeWGU259JnaQ34Inuec4R38JCyBx2edgD0=
+github.com/karrick/godirwalk v1.8.0/go.mod h1:H5KPZjojv4lE+QYImBI8xVtrBRgYrIVsaRPx4tDPEn4=
+github.com/karrick/godirwalk v1.10.3/go.mod h1:RoGL9dQei4vP9ilrpETWE8CLOZ1kiN0LhBygSwrAsHA=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
+github.com/klauspost/compress v1.9.5 h1:U+CaK85mrNNb4k8BNOfgJtJ/gr6kswUCFj6miSzVC6M=
+github.com/klauspost/compress v1.9.5/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/koron/go-ssdp v0.0.0-20180514024734-4a0ed625a78b h1:wxtKgYHEncAU00muMD06dzLiahtGM1eouRNOzVV7tdQ=
 github.com/koron/go-ssdp v0.0.0-20180514024734-4a0ed625a78b/go.mod h1:5Ky9EC2xfoUKUor0Hjgi2BJhCSXJfMOFlmyYrVKGQMk=
 github.com/koron/go-ssdp v0.0.0-20191105050749-2e1c40ed0b5d h1:68u9r4wEvL3gYg2jvAOgROwZ3H+Y3hIDk4tbbmIjcYQ=
@@ -720,6 +763,8 @@ github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czP
 github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329 h1:2gxZ0XQIU/5z3Z3bUBu+FXuk2pFbkN6tcwi/pjyaDic=
 github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190312143242-1de009706dbe/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
+github.com/markbates/oncer v0.0.0-20181203154359-bf2de49a0be2/go.mod h1:Ld9puTsIW75CHf65OeIOkyKbteujpZVXDpWK6YGZbxE=
+github.com/markbates/safe v1.0.1/go.mod h1:nAqgmRi7cY2nqMc92/bSEeQA+R4OheNU2T1kNSCBdG0=
 github.com/marten-seemann/qpack v0.1.0/go.mod h1:LFt1NU/Ptjip0C2CPkhimBz5CGE3WGDAUWqna+CNTrI=
 github.com/marten-seemann/qtls v0.9.1/go.mod h1:T1MmAdDPyISzxlK6kjRr0pcZFBVd1OZbBb/j3cvzHhk=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
@@ -750,6 +795,7 @@ github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrk
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
 github.com/mr-tron/base58 v1.1.0/go.mod h1:xcD2VGqlgYjBdcBLw+TuYLr8afG+Hj8g2eTVqeSzSU8=
 github.com/mr-tron/base58 v1.1.1/go.mod h1:xcD2VGqlgYjBdcBLw+TuYLr8afG+Hj8g2eTVqeSzSU8=
 github.com/mr-tron/base58 v1.1.2/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
@@ -838,8 +884,10 @@ github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYr
 github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
 github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
+github.com/pelletier/go-toml v1.4.0/go.mod h1:PN7xzY2wHTK0K9p34ErDQMlFxa51Fk0OUruD3k1mMwo=
 github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2 h1:JhzVVoYvbOACxoUmOs6V/G4D5nPVUW73rKvXxP4XUJc=
 github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2/go.mod h1:iIss55rKnNBTvrwdmkUpLnDpZoAHvWaiq5+iMmen4AE=
+github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -854,6 +902,8 @@ github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
 github.com/prometheus/procfs v0.0.0-20180725123919-05ee40e3a273/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
+github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
 github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
@@ -882,6 +932,9 @@ github.com/shurcooL/sanitized_anchor_name v0.0.0-20170918181015-86672fcb3f95/go.
 github.com/shurcooL/users v0.0.0-20180125191416-49c67e49c537/go.mod h1:QJTqeLYEDaXHZDBsXlPCDqdhQuJkuw4NOtaxYe3xii4=
 github.com/shurcooL/webdavfs v0.0.0-20170829043945-18c3829fa133/go.mod h1:hKmq5kWdCj2z2KEozexVbfEZIWiTjhE0+UjmZgPqehw=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/sirupsen/logrus v1.4.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
+github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/assertions v1.0.0 h1:UVQPSSmc3qtTi+zPPkCXvZX9VvW/xT/NsRvKfwY81a8=
@@ -901,6 +954,7 @@ github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0b
 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
+github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
 github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
@@ -919,10 +973,12 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE=
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/tarm/serial v0.0.0-20180830185346-98f6abe2eb07/go.mod h1:kDXzergiv9cbyO7IOYJZWg1U88JhDg3PB6klq9Hg2pA=
-github.com/textileio/go-datastore v0.4.5-0.20200819232101-baa577bf9422 h1:DNpznzcn7pd1Cn2fdKHfPPw62r8Ii4DGVnQRu9Jbvok=
-github.com/textileio/go-datastore v0.4.5-0.20200819232101-baa577bf9422/go.mod h1:/38mp5DMgxCZrb5wpgPlWGXuZ99/ur8wgHDSXU5zCjU=
-github.com/textileio/go-ds-badger v0.2.5-0.20200819232634-de89720b5d6a h1:AdjNdwIWrZAUrXfC9IHG8eKhRlJarXZoN9CmGlfTeLA=
-github.com/textileio/go-ds-badger v0.2.5-0.20200819232634-de89720b5d6a/go.mod h1:0kLVpG7eeM95s4rS78lQe4eG5DCk+cnU8xas2nPSdZY=
+github.com/textileio/go-datastore-extensions v1.0.1 h1:qIJGqJaigQ1wD4TdwS/hf73u0HChhXvvUSJuxBEKS+c=
+github.com/textileio/go-datastore-extensions v1.0.1/go.mod h1:Pzj9FDRkb55910dr/FX8M7WywvnS26gBgEDez1ZBuLE=
+github.com/textileio/go-ds-badger v0.2.7-0.20201204225019-4ee78c4a40e2 h1:VIWkccJp5k8x1hbwQsSn6Xnns4wUDEW6XdiMLiOpr7k=
+github.com/textileio/go-ds-badger v0.2.7-0.20201204225019-4ee78c4a40e2/go.mod h1:qEZ/z1KyoRhGS5MYEbIcWUCCPd/0HxCkFDVeJgP1RcI=
+github.com/textileio/go-ds-mongo v0.1.4-0.20201205032204-63f494d9c3f3 h1:agZyjIQP6gqAcKxp06bAM39iAYd84kfkGJ8XyY3hNQI=
+github.com/textileio/go-ds-mongo v0.1.4-0.20201205032204-63f494d9c3f3/go.mod h1:Zf6JlMPiIQUUmGlFFn5Z65C9p9LAvPg7XvX+qdGmTsU=
 github.com/tidwall/gjson v1.3.5 h1:2oW9FBNu8qt9jy5URgrzsVx/T/KSn3qn/smJQ0crlDQ=
 github.com/tidwall/gjson v1.3.5/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
 github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
@@ -962,6 +1018,10 @@ github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee h1:lYbXeSv
 github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee/go.mod h1:m2aV4LZI4Aez7dP5PMyVKEHhUyEJ/RjmPEDOpDvudHg=
 github.com/whyrusleeping/yamux v1.1.5/go.mod h1:E8LnQQ8HKx5KD29HZFUwM1PxCOdPRzGwur1mcYhXcD8=
 github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
+github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
+github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
+github.com/xdg/stringprep v0.0.0-20180714160509-73f8eece6fdc h1:n+nNi93yXLkJvKwXNP9d55HC7lGK4H/SRcwB5IaUZLo=
+github.com/xdg/stringprep v0.0.0-20180714160509-73f8eece6fdc/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo=
 github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
@@ -970,6 +1030,8 @@ github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:
 github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
+go.mongodb.org/mongo-driver v1.4.0 h1:C8rFn1VF4GVEM/rG+dSoMmlm2pyQ9cs2/oRtUATejRU=
+go.mongodb.org/mongo-driver v1.4.0/go.mod h1:llVBH2pkj9HywK0Dtdt6lDikOjFLbceHVu/Rc0iMKLs=
 go.opencensus.io v0.18.0/go.mod h1:vKdFvxhtzZ9onBp9VKHK8z/sRpBMnKAsufL7wlDrCOA=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.1/go.mod h1:Ap50jQcDJrx6rB6VgeeFPtuPIf3wMRvRfrfYDO6+BmA=
@@ -1002,6 +1064,7 @@ golang.org/x/crypto v0.0.0-20190225124518-7f87c0fbb88b/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20190228161510-8dd112bcdc25/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190313024323-a1f597ede03a/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20190422162423-af44ce270edf/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE=
 golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190513172903-22d7a77e9e5f/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -1048,6 +1111,7 @@ golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR
 golang.org/x/net v0.0.0-20190611141213-3f473d35a33a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200421231249-e086a090c8fd h1:QPwSajcTUrFriMF1nJ3XzgoqakqQEsnZf9LdXdi2nkI=
 golang.org/x/net v0.0.0-20200421231249-e086a090c8fd/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -1059,7 +1123,9 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190412183630-56d357773e84/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a h1:WXEvlFVvvGxCJLG6REjsT03iWnKLEWinaScsxF2Vm2o=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -1074,12 +1140,15 @@ golang.org/x/sys v0.0.0-20190228124157-a34e9553db1e/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190302025703-b6889370fb10/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190316082340-a2f829d7f35f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190405154228-4b34438f7a67/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190419153524-e8e3143a4f4a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190524122548-abf6ff778158/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190526052359-791d8a0f4d09/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190531175056-4c3a928424d2/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190610200419-93c9922d18ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1110,8 +1179,12 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20190329151228-23e29df326fe/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20190416151739-9c9e1878f421/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20190420181800-aa740d480789/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/tools v0.0.0-20190531172133-b3315ee88b7d/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
diff --git a/jsonpatcher/jsonpatcher.go b/jsonpatcher/jsonpatcher.go
index 0e075c5..cfba615 100644
--- a/jsonpatcher/jsonpatcher.go
+++ b/jsonpatcher/jsonpatcher.go
@@ -10,11 +10,11 @@ import (
 	"time"
 
 	jsonpatch "github.com/evanphx/json-patch"
+	ds "github.com/ipfs/go-datastore"
 	cbornode "github.com/ipfs/go-ipld-cbor"
 	format "github.com/ipfs/go-ipld-format"
 	logging "github.com/ipfs/go-log"
 	"github.com/multiformats/go-multihash"
-	ds "github.com/textileio/go-datastore"
 	core "github.com/textileio/go-threads/core/db"
 )
 
@@ -40,7 +40,6 @@ func (ot operationType) String() (s string) {
 
 var (
 	log                           = logging.Logger("jsonpatcher")
-	errSavingNonExistentInstance  = errors.New("can't save nonexistent instance")
 	errCantCreateExistingInstance = errors.New("cant't create already existent instance")
 	errUnknownOperation           = errors.New("unknown operation type")
 )
@@ -104,8 +103,13 @@ func (jp *jsonPatcher) Create(actions []core.Action) ([]core.Event, format.Node,
 	return events, n, nil
 }
 
-func (jp *jsonPatcher) Reduce(events []core.Event, datastore ds.TxnDatastore, baseKey ds.Key, indexFunc core.IndexFunc) ([]core.ReduceAction, error) {
-	txn, err := datastore.NewTransaction(false)
+func (jp *jsonPatcher) Reduce(
+	events []core.Event,
+	store ds.TxnDatastore,
+	baseKey ds.Key,
+	indexFunc core.IndexFunc,
+) ([]core.ReduceAction, error) {
+	txn, err := store.NewTransaction(false)
 	if err != nil {
 		return nil, err
 	}
diff --git a/net/api/service.go b/net/api/service.go
index 5644a53..bbf6329 100644
--- a/net/api/service.go
+++ b/net/api/service.go
@@ -30,7 +30,7 @@ type Service struct {
 	net net.Net
 }
 
-// Config specifies server settings.
+// Config specifies service settings.
 type Config struct {
 	Debug bool
 }
diff --git a/net/net.go b/net/net.go
index b3640d6..b863758 100644
--- a/net/net.go
+++ b/net/net.go
@@ -226,7 +226,11 @@ func (n *net) GetToken(ctx context.Context, identity thread.Identity) (tok threa
 	return thread.NewToken(n.getPrivKey(), key)
 }
 
-func (n *net) CreateThread(_ context.Context, id thread.ID, opts ...core.NewThreadOption) (info thread.Info, err error) {
+func (n *net) CreateThread(
+	_ context.Context,
+	id thread.ID,
+	opts ...core.NewThreadOption,
+) (info thread.Info, err error) {
 	args := &core.NewThreadOptions{}
 	for _, opt := range opts {
 		opt(args)
@@ -267,7 +271,11 @@ func (n *net) CreateThread(_ context.Context, id thread.ID, opts ...core.NewThre
 	return n.getThreadWithAddrs(id)
 }
 
-func (n *net) AddThread(ctx context.Context, addr ma.Multiaddr, opts ...core.NewThreadOption) (info thread.Info, err error) {
+func (n *net) AddThread(
+	ctx context.Context,
+	addr ma.Multiaddr,
+	opts ...core.NewThreadOption,
+) (info thread.Info, err error) {
 	args := &core.NewThreadOptions{}
 	for _, opt := range opts {
 		opt(args)
@@ -480,7 +488,12 @@ func (n *net) deleteThread(ctx context.Context, id thread.ID) error {
 	return n.store.DeleteThread(id) // Delete logstore keys, addresses, heads, and metadata
 }
 
-func (n *net) AddReplicator(ctx context.Context, id thread.ID, paddr ma.Multiaddr, opts ...core.ThreadOption) (pid peer.ID, err error) {
+func (n *net) AddReplicator(
+	ctx context.Context,
+	id thread.ID,
+	paddr ma.Multiaddr,
+	opts ...core.ThreadOption,
+) (pid peer.ID, err error) {
 	args := &core.ThreadOptions{}
 	for _, opt := range opts {
 		opt(args)
@@ -607,7 +620,12 @@ func getDialable(addr ma.Multiaddr) (ma.Multiaddr, error) {
 	return ma.NewMultiaddr(parts[0])
 }
 
-func (n *net) CreateRecord(ctx context.Context, id thread.ID, body format.Node, opts ...core.ThreadOption) (tr core.ThreadRecord, err error) {
+func (n *net) CreateRecord(
+	ctx context.Context,
+	id thread.ID,
+	body format.Node,
+	opts ...core.ThreadOption,
+) (tr core.ThreadRecord, err error) {
 	args := &core.ThreadOptions{}
 	for _, opt := range opts {
 		opt(args)
@@ -650,7 +668,13 @@ func (n *net) CreateRecord(ctx context.Context, id thread.ID, body format.Node,
 	return tr, nil
 }
 
-func (n *net) AddRecord(ctx context.Context, id thread.ID, lid peer.ID, rec core.Record, opts ...core.ThreadOption) error {
+func (n *net) AddRecord(
+	ctx context.Context,
+	id thread.ID,
+	lid peer.ID,
+	rec core.Record,
+	opts ...core.ThreadOption,
+) error {
 	args := &core.ThreadOptions{}
 	for _, opt := range opts {
 		opt(args)
@@ -684,7 +708,12 @@ func (n *net) AddRecord(ctx context.Context, id thread.ID, lid peer.ID, rec core
 	return n.server.pushRecord(ctx, id, lid, rec)
 }
 
-func (n *net) GetRecord(ctx context.Context, id thread.ID, rid cid.Cid, opts ...core.ThreadOption) (core.Record, error) {
+func (n *net) GetRecord(
+	ctx context.Context,
+	id thread.ID,
+	rid cid.Cid,
+	opts ...core.ThreadOption,
+) (core.Record, error) {
 	args := &core.ThreadOptions{}
 	for _, opt := range opts {
 		opt(args)
@@ -1022,7 +1051,13 @@ func (n *net) currentHead(tid thread.ID, lid peer.ID) (cid.Cid, error) {
 }
 
 // newRecord creates a new record with the given body as a new event body.
-func (n *net) newRecord(ctx context.Context, id thread.ID, lg thread.LogInfo, body format.Node, pk thread.PubKey) (core.Record, error) {
+func (n *net) newRecord(
+	ctx context.Context,
+	id thread.ID,
+	lg thread.LogInfo,
+	body format.Node,
+	pk thread.PubKey,
+) (core.Record, error) {
 	if lg.PrivKey == nil {
 		return nil, fmt.Errorf("a private-key is required to create records")
 	}
@@ -1062,7 +1097,13 @@ func (n *net) getPrivKey() crypto.PrivKey {
 // offset but not farther than limit.
 // It is possible to reach limit before offset, meaning that the caller
 // will be responsible for the remaining traversal.
-func (n *net) getLocalRecords(ctx context.Context, id thread.ID, lid peer.ID, offset cid.Cid, limit int) ([]core.Record, error) {
+func (n *net) getLocalRecords(
+	ctx context.Context,
+	id thread.ID,
+	lid peer.ID,
+	offset cid.Cid,
+	limit int,
+) ([]core.Record, error) {
 	lg, err := n.store.GetLog(id, lid)
 	if err != nil {
 		return nil, err
@@ -1245,7 +1286,13 @@ func (n *net) getOrCreateLog(id thread.ID, identity thread.PubKey) (info thread.
 
 // createExternalLogIfNotExist creates an external log if doesn't exists. The created
 // log will have cid.Undef as the current head. Is thread-safe.
-func (n *net) createExternalLogIfNotExist(tid thread.ID, lid peer.ID, pubKey crypto.PubKey, privKey crypto.PrivKey, addrs []ma.Multiaddr) error {
+func (n *net) createExternalLogIfNotExist(
+	tid thread.ID,
+	lid peer.ID,
+	pubKey crypto.PubKey,
+	privKey crypto.PrivKey,
+	addrs []ma.Multiaddr,
+) error {
 	ts := n.semaphores.Get(semaThreadUpdate(tid))
 	ts.Acquire()
 	defer ts.Release()
@@ -1365,7 +1412,6 @@ func (n *net) threadOffsets(tid thread.ID) (map[peer.ID]cid.Cid, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	var offsets = make(map[peer.ID]cid.Cid, len(info.Logs))
 	for _, lg := range info.Logs {
 		var has bool
@@ -1375,13 +1421,11 @@ func (n *net) threadOffsets(tid thread.ID) (map[peer.ID]cid.Cid, error) {
 				return nil, err
 			}
 		}
-
 		if has {
 			offsets[lg.ID] = lg.Head
 		} else {
 			offsets[lg.ID] = cid.Undef
 		}
 	}
-
 	return offsets, nil
 }
diff --git a/threadsd/main.go b/threadsd/main.go
index 05e25ed..29decad 100644
--- a/threadsd/main.go
+++ b/threadsd/main.go
@@ -6,7 +6,9 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"net/url"
 	"os"
+	"os/signal"
 	"time"
 
 	"github.com/improbable-eng/grpc-web/go/grpcweb"
@@ -14,9 +16,11 @@ import (
 	connmgr "github.com/libp2p/go-libp2p-connmgr"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/namsral/flag"
+	mongods "github.com/textileio/go-ds-mongo"
 	"github.com/textileio/go-threads/api"
 	pb "github.com/textileio/go-threads/api/pb"
 	"github.com/textileio/go-threads/common"
+	kt "github.com/textileio/go-threads/db/keytransform"
 	netapi "github.com/textileio/go-threads/net/api"
 	netpb "github.com/textileio/go-threads/net/api/pb"
 	"github.com/textileio/go-threads/util"
@@ -37,6 +41,9 @@ func main() {
 	connGracePeriod := fs.Duration("connGracePeriod", time.Second*20, "Duration a new opened connection is not subject to pruning")
 	keepAliveInterval := fs.Duration("keepAliveInterval", time.Second*5, "Websocket keepalive interval (must be >= 1s)")
 	enableNetPubsub := fs.Bool("enableNetPubsub", false, "Enables thread networking over libp2p pubsub")
+	mongoUri := fs.String("mongoUri", "", "MongoDB URI (if not provided, an embedded Badger datastore will be used)")
+	mongoDatabase := fs.String("mongoDatabase", "", "MongoDB database name (required with mongoUri")
+	badgerLowMem := fs.Bool("badgerLowMem", false, "Use Badger's low memory settings")
 	debug := fs.Bool("debug", false, "Enables debug logging")
 	if err := fs.Parse(os.Args[1:]); err != nil {
 		log.Fatal(err)
@@ -55,6 +62,19 @@ func main() {
 		log.Fatal(err)
 	}
 
+	var parsedMongoUri *url.URL
+	if len(*mongoUri) != 0 {
+		parsedMongoUri, err = url.Parse(*mongoUri)
+		if err != nil {
+			log.Fatalf("parsing mongoUri: %v", err)
+		}
+		if len(*mongoDatabase) == 0 {
+			log.Fatal("mongoDatabase is required with mongoUri")
+		}
+	} else {
+		log.Debugf("badgerLowMem: %v", *badgerLowMem)
+	}
+
 	if err := util.SetupDefaultLoggingConfig(*repo); err != nil {
 		log.Fatal(err)
 	}
@@ -73,23 +93,45 @@ func main() {
 	log.Debugf("connGracePeriod: %v", *connGracePeriod)
 	log.Debugf("keepAliveInterval: %v", *keepAliveInterval)
 	log.Debugf("enableNetPubsub: %v", *enableNetPubsub)
+	if parsedMongoUri != nil {
+		log.Debugf("mongoUri: %v", parsedMongoUri.Redacted())
+		log.Debugf("mongoDatabase: %v", *mongoDatabase)
+	} else {
+		log.Debugf("badgerLowMem: %v", *badgerLowMem)
+	}
 	log.Debugf("debug: %v", *debug)
 
-	n, err := common.DefaultNetwork(
-		*repo,
+	opts := []common.NetOption{
 		common.WithNetHostAddr(hostAddr),
 		common.WithConnectionManager(connmgr.NewConnManager(*connLowWater, *connHighWater, *connGracePeriod)),
 		common.WithNetPubSub(*enableNetPubsub),
-		common.WithNetDebug(*debug))
+		common.WithNetDebug(*debug),
+	}
+	if parsedMongoUri != nil {
+		opts = append(opts, common.WithNetMongoPersistence(*mongoUri, *mongoDatabase))
+	} else {
+		opts = append(opts, common.WithNetBadgerPersistence(*repo))
+	}
+	n, err := common.DefaultNetwork(opts...)
 	if err != nil {
 		log.Fatal(err)
 	}
 	defer n.Close()
 	n.Bootstrap(util.DefaultBoostrapPeers())
 
-	service, err := api.NewService(n, api.Config{
-		RepoPath: *repo,
-		Debug:    *debug,
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var store kt.TxnDatastoreExtended
+	if *mongoUri != "" {
+		store, err = mongods.New(ctx, *mongoUri, *mongoDatabase, mongods.WithCollName("eventstore"))
+	} else {
+		store, err = util.NewBadgerDatastore(*repo, *badgerLowMem)
+	}
+	if err != nil {
+		log.Fatal(err)
+	}
+	service, err := api.NewService(store, n, api.Config{
+		Debug: *debug,
 	})
 	if err != nil {
 		log.Fatal(err)
@@ -148,7 +190,10 @@ func main() {
 		}
 	}()
 
-	defer func() {
+	fmt.Println("Welcome to Threads!")
+	fmt.Println("Your peer ID is " + n.Host().ID().String())
+
+	handleInterrupt(func() {
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
 		if err := proxy.Shutdown(ctx); err != nil {
@@ -158,12 +203,14 @@ func main() {
 		if err := n.Close(); err != nil {
 			log.Fatal(err)
 		}
-	}()
-
-	fmt.Println("Welcome to Threads!")
-	fmt.Println("Your peer ID is " + n.Host().ID().String())
-
-	log.Debug("threadsd started")
+	})
+}
 
-	select {}
+func handleInterrupt(stop func()) {
+	quit := make(chan os.Signal)
+	signal.Notify(quit, os.Interrupt)
+	<-quit
+	fmt.Println("Gracefully stopping... (press Ctrl+C again to force)")
+	stop()
+	os.Exit(1)
 }
diff --git a/util/util.go b/util/util.go
index 575d091..b28e270 100644
--- a/util/util.go
+++ b/util/util.go
@@ -4,37 +4,51 @@ import (
 	"crypto/rand"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"path/filepath"
-	"strings"
 
 	"github.com/alecthomas/jsonschema"
+	"github.com/dgraph-io/badger/options"
 	ipfslite "github.com/hsanjuan/ipfs-lite"
 	logging "github.com/ipfs/go-log/v2"
-	"github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/peer"
-	swarm "github.com/libp2p/go-libp2p-swarm"
 	ma "github.com/multiformats/go-multiaddr"
+	mbase "github.com/multiformats/go-multibase"
 	"github.com/phayes/freeport"
+	badger "github.com/textileio/go-ds-badger"
 	core "github.com/textileio/go-threads/core/db"
+	kt "github.com/textileio/go-threads/db/keytransform"
 	"github.com/tidwall/sjson"
 	"go.uber.org/zap/zapcore"
 )
 
 var (
 	bootstrapPeers = []string{
-		"/ip4/104.210.43.77/tcp/4001/p2p/QmR69wtWUMm1TWnmuD4JqC1TWLZcc8iR2KrTenfZZbiztd",       // us-west-1 hub ipfs host
-		"/ip4/104.210.43.77/tcp/4006/p2p/12D3KooWQEtCBXMKjVas6Ph1pUHG2T4Lc9j1KvnAipojP2xcKU7n", // us-west-1 hub threads host
+		// hub-production ipfs-hub-0 host
+		"/ip4/40.76.153.74/tcp/4001/p2p/QmR69wtWUMm1TWnmuD4JqC1TWLZcc8iR2KrTenfZZbiztd",
+		// hub-production hub-0 threads host
+		"/ip4/52.186.99.239/tcp/4006/p2p/12D3KooWQEtCBXMKjVas6Ph1pUHG2T4Lc9j1KvnAipojP2xcKU7n",
+		// go-ipfs 0.7.0 default bootstrap peers
+		"/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
+		"/dnsaddr/bootstrap.libp2p.io/p2p/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
+		"/dnsaddr/bootstrap.libp2p.io/p2p/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
+		"/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
+		"/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ",
+		"/ip4/104.131.131.82/udp/4001/quic/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ",
 	}
 )
 
-// CanDial returns whether or not the address is dialable.
-func CanDial(addr ma.Multiaddr, s *swarm.Swarm) bool {
-	parts := strings.Split(addr.String(), "/"+ma.ProtocolWithCode(ma.P_P2P).Name)
-	addr, _ = ma.NewMultiaddr(parts[0])
-	tr := s.TransportForDialing(addr)
-	return tr != nil && tr.CanDial(addr)
+// NewBadgerDatastore returns a badger based datastore.
+func NewBadgerDatastore(repoPath string, lowMem bool) (kt.TxnDatastoreExtended, error) {
+	path := filepath.Join(repoPath, "eventstore")
+	if err := os.MkdirAll(path, os.ModePerm); err != nil {
+		return nil, err
+	}
+	opts := badger.DefaultOptions
+	if lowMem {
+		opts.TableLoadingMode = options.FileIO
+	}
+	return badger.NewDatastore(path, &opts)
 }
 
 // SetupDefaultLoggingConfig sets up a standard logging configuration.
@@ -72,36 +86,6 @@ func SetLogLevels(systems map[string]logging.LogLevel) error {
 	return nil
 }
 
-func LoadKey(pth string) crypto.PrivKey {
-	var priv crypto.PrivKey
-	_, err := os.Stat(pth)
-	if os.IsNotExist(err) {
-		priv, _, err = crypto.GenerateKeyPair(crypto.Ed25519, 0)
-		if err != nil {
-			panic(err)
-		}
-		key, err := crypto.MarshalPrivateKey(priv)
-		if err != nil {
-			panic(err)
-		}
-		if err = ioutil.WriteFile(pth, key, 0400); err != nil {
-			panic(err)
-		}
-	} else if err != nil {
-		panic(err)
-	} else {
-		key, err := ioutil.ReadFile(pth)
-		if err != nil {
-			panic(err)
-		}
-		priv, err = crypto.UnmarshalPrivateKey(key)
-		if err != nil {
-			panic(err)
-		}
-	}
-	return priv
-}
-
 func DefaultBoostrapPeers() []peer.AddrInfo {
 	ipfspeers := ipfslite.DefaultBootstrapPeers()
 	textilepeers, err := ParseBootstrapPeers(bootstrapPeers)
@@ -203,3 +187,12 @@ func GenerateRandomBytes(n int) []byte {
 	}
 	return b
 }
+
+func MakeToken(n int) string {
+	bytes := GenerateRandomBytes(n)
+	encoded, err := mbase.Encode(mbase.Base32, bytes)
+	if err != nil {
+		panic(err)
+	}
+	return encoded
+}


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index b69b582..962f47e 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -15,7 +15,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v1
         with:
-          go-version: 1.14
+          go-version: 1.15
       - name: Check out code
         uses: actions/checkout@v1
       - name: Cache dependencies
@@ -40,7 +40,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v1
         with:
-          go-version: 1.14
+          go-version: 1.15
       - name: Check out code
         uses: actions/checkout@v1
       - name: Cache dependencies
diff --git a/api/client/client_test.go b/api/client/client_test.go
index 2ba24d9..671229c 100644
--- a/api/client/client_test.go
+++ b/api/client/client_test.go
@@ -7,7 +7,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"log"
 	"math/rand"
 	"net"
@@ -20,6 +19,8 @@ import (
 	"github.com/libp2p/go-libp2p-core/crypto"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/phayes/freeport"
+	mongods "github.com/textileio/go-ds-mongo"
+	"github.com/textileio/go-ds-mongo/test"
 	"github.com/textileio/go-threads/api"
 	. "github.com/textileio/go-threads/api/client"
 	pb "github.com/textileio/go-threads/api/pb"
@@ -30,6 +31,13 @@ import (
 	"google.golang.org/grpc"
 )
 
+func TestMain(m *testing.M) {
+	cleanup := test.StartMongoDB()
+	exitVal := m.Run()
+	cleanup()
+	os.Exit(exitVal)
+}
+
 func TestClient_GetToken(t *testing.T) {
 	t.Parallel()
 	client, done := setup(t)
@@ -182,7 +190,11 @@ func TestClient_NewCollection(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		if err != nil {
 			t.Fatalf("failed to add new collection: %v", err)
 		}
@@ -197,7 +209,11 @@ func TestClient_UpdateCollection(t *testing.T) {
 	id := thread.NewIDV1(thread.Raw, 32)
 	err := client.NewDB(context.Background(), id)
 	checkErr(t, err)
-	err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+	err = client.NewCollection(
+		context.Background(),
+		id,
+		db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+	)
 	checkErr(t, err)
 
 	t.Run("test update collection", func(t *testing.T) {
@@ -226,7 +242,11 @@ func TestClient_DeleteCollection(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		err = client.DeleteCollection(context.Background(), id, collectionName)
@@ -348,7 +368,11 @@ func TestClient_Create(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		_, err = client.Create(context.Background(), id, collectionName, Instances{createPerson()})
@@ -361,7 +385,11 @@ func TestClient_Create(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		ids, err := client.Create(context.Background(), id, collectionName, Instances{&PersonWithoutID{
@@ -434,7 +462,11 @@ func TestClient_Save(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		person := createPerson()
@@ -460,7 +492,11 @@ func TestClient_Delete(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		person := createPerson()
@@ -486,7 +522,11 @@ func TestClient_Has(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		person := createPerson()
@@ -515,7 +555,11 @@ func TestClient_Find(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		person := createPerson()
@@ -591,7 +635,11 @@ func TestClient_FindByID(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		person := createPerson()
@@ -621,7 +669,11 @@ func TestClient_ReadTransaction(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		person := createPerson()
@@ -688,7 +740,11 @@ func TestClient_WriteTransaction(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		existingPerson := createPerson()
@@ -774,7 +830,11 @@ func TestClient_WriteTransaction(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		existingPerson := createPerson()
@@ -823,7 +883,11 @@ func TestClient_Listen(t *testing.T) {
 		id := thread.NewIDV1(thread.Raw, 32)
 		err := client.NewDB(context.Background(), id)
 		checkErr(t, err)
-		err = client.NewCollection(context.Background(), id, db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)})
+		err = client.NewCollection(
+			context.Background(),
+			id,
+			db.CollectionConfig{Name: collectionName, Schema: util.SchemaFromSchemaString(schema)},
+		)
 		checkErr(t, err)
 
 		person := createPerson()
@@ -931,22 +995,27 @@ func setup(t *testing.T) (*Client, func()) {
 
 func makeServer(t *testing.T) (ma.Multiaddr, func()) {
 	time.Sleep(time.Second * time.Duration(rand.Intn(5)))
-	dir, err := ioutil.TempDir("", "")
-	if err != nil {
-		t.Fatal(err)
-	}
 	n, err := common.DefaultNetwork(
-		dir,
+		common.WithNetMongoPersistence(test.MongoUri, util.MakeToken(12)),
 		common.WithNetHostAddr(util.FreeLocalAddr()),
 		common.WithNetPubSub(true),
-		common.WithNetDebug(true))
+		common.WithNetDebug(true),
+	)
 	if err != nil {
 		t.Fatal(err)
 	}
-	n.Bootstrap(util.DefaultBoostrapPeers())
-	service, err := api.NewService(n, api.Config{
-		RepoPath: dir,
-		Debug:    true,
+	ctx, cancel := context.WithCancel(context.Background())
+	store, err := mongods.New(
+		ctx,
+		test.MongoUri,
+		util.MakeToken(12),
+		mongods.WithCollName("eventstore"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	service, err := api.NewService(store, n, api.Config{
+		Debug: true,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -978,7 +1047,10 @@ func makeServer(t *testing.T) (ma.Multiaddr, func()) {
 		if err := n.Close(); err != nil {
 			t.Fatal(err)
 		}
-		_ = os.RemoveAll(dir)
+		if err := store.Close(); err != nil {
+			t.Fatal(err)
+		}
+		cancel()
 	}
 }
 
diff --git a/db/bench_test.go b/db/bench_test.go
index 51809c1..5501edc 100644
--- a/db/bench_test.go
+++ b/db/bench_test.go
@@ -60,15 +60,23 @@ func checkBenchErr(b *testing.B, err error) {
 func createBenchDB(b *testing.B, opts ...NewOption) (*DB, func()) {
 	dir, err := ioutil.TempDir("", "")
 	checkBenchErr(b, err)
-	n, err := common.DefaultNetwork(dir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(dir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkBenchErr(b, err)
-	opts = append(opts, WithNewRepoPath(dir))
-	d, err := NewDB(context.Background(), n, thread.NewIDV1(thread.Raw, 32), opts...)
+	store, err := util.NewBadgerDatastore(dir, false)
+	checkBenchErr(b, err)
+	d, err := NewDB(context.Background(), store, n, thread.NewIDV1(thread.Raw, 32), opts...)
 	checkBenchErr(b, err)
 	return d, func() {
 		if err := n.Close(); err != nil {
 			panic(err)
 		}
+		if err := store.Close(); err != nil {
+			panic(err)
+		}
 		_ = os.RemoveAll(dir)
 	}
 }
diff --git a/db/collection_test.go b/db/collection_test.go
index 95355fa..51fd005 100644
--- a/db/collection_test.go
+++ b/db/collection_test.go
@@ -852,7 +852,7 @@ func TestGetInstance(t *testing.T) {
 
 func TestModifiedSince(t *testing.T) {
 	t.Parallel()
-	t.Run("WithiSingleCreate", func(t *testing.T) {
+	t.Run("WithSingleCreate", func(t *testing.T) {
 		t.Parallel()
 		db, clean := createTestDB(t)
 		defer clean()
diff --git a/db/db_test.go b/db/db_test.go
index 5b3ad2c..7248d62 100644
--- a/db/db_test.go
+++ b/db/db_test.go
@@ -11,9 +11,9 @@ import (
 	"testing"
 	"time"
 
+	ds "github.com/ipfs/go-datastore"
 	format "github.com/ipfs/go-ipld-format"
 	"github.com/multiformats/go-multiaddr"
-	ds "github.com/textileio/go-datastore"
 	"github.com/textileio/go-threads/common"
 	"github.com/textileio/go-threads/core/app"
 	core "github.com/textileio/go-threads/core/db"
@@ -29,12 +29,20 @@ func TestE2EWithThreads(t *testing.T) {
 	checkErr(t, err)
 	defer os.RemoveAll(tmpDir1)
 
-	n1, err := common.DefaultNetwork(tmpDir1, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n1, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(tmpDir1),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkErr(t, err)
 	defer n1.Close()
 
+	store, err := util.NewBadgerDatastore(tmpDir1, false)
+	checkErr(t, err)
+	defer store.Close()
+
 	id1 := thread.NewIDV1(thread.Raw, 32)
-	d1, err := NewDB(context.Background(), n1, id1, WithNewRepoPath(tmpDir1))
+	d1, err := NewDB(context.Background(), store, n1, id1)
 	checkErr(t, err)
 	defer d1.Close()
 	c1, err := d1.NewCollection(CollectionConfig{
@@ -69,7 +77,11 @@ func TestE2EWithThreads(t *testing.T) {
 	tmpDir2, err := ioutil.TempDir("", "")
 	checkErr(t, err)
 	defer os.RemoveAll(tmpDir2)
-	n2, err := common.DefaultNetwork(tmpDir2, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n2, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(tmpDir2),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkErr(t, err)
 	defer n2.Close()
 
@@ -79,7 +91,19 @@ func TestE2EWithThreads(t *testing.T) {
 		Name:   "dummy",
 		Schema: util.SchemaFromInstance(&dummy{}, false),
 	}
-	d2, err := NewDBFromAddr(context.Background(), n2, addr, ti.Key, WithNewRepoPath(tmpDir2), WithNewCollections(cc))
+
+	store2, err := util.NewBadgerDatastore(tmpDir2, false)
+	checkErr(t, err)
+	defer store2.Close()
+
+	d2, err := NewDBFromAddr(
+		context.Background(),
+		store2,
+		n2,
+		addr,
+		ti.Key,
+		WithNewCollections(cc),
+	)
 	checkErr(t, err)
 	defer d2.Close()
 	c2 := d1.GetCollection("dummy")
@@ -107,12 +131,20 @@ func TestMissingCollection(t *testing.T) {
 	checkErr(t, err)
 	defer os.RemoveAll(tmpDir)
 
-	n, err := common.DefaultNetwork(tmpDir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(tmpDir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkErr(t, err)
 	defer n.Close()
 
+	store, err := util.NewBadgerDatastore(tmpDir, false)
+	checkErr(t, err)
+	defer store.Close()
+
 	id := thread.NewIDV1(thread.Raw, 32)
-	db, err := NewDB(context.Background(), n, id, WithNewRepoPath(tmpDir))
+	db, err := NewDB(context.Background(), store, n, id)
 	checkErr(t, err)
 	defer db.Close()
 	c, err := db.NewCollection(CollectionConfig{
@@ -136,12 +168,20 @@ func TestWithNewName(t *testing.T) {
 	checkErr(t, err)
 	defer os.RemoveAll(tmpDir)
 
-	n, err := common.DefaultNetwork(tmpDir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(tmpDir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
+	checkErr(t, err)
+
+	store, err := util.NewBadgerDatastore(tmpDir, false)
 	checkErr(t, err)
+	defer store.Close()
 
 	name := "my-db"
 	id := thread.NewIDV1(thread.Raw, 32)
-	d, err := NewDB(context.Background(), n, id, WithNewRepoPath(tmpDir), WithNewName(name))
+	d, err := NewDB(context.Background(), store, n, id, WithNewName(name))
 	checkErr(t, err)
 	if d.name != name {
 		t.Fatalf("expected name %s, got %s", name, d.name)
@@ -155,11 +195,15 @@ func TestWithNewName(t *testing.T) {
 	checkErr(t, d.Close())
 
 	time.Sleep(time.Second * 3)
-	n, err = common.DefaultNetwork(tmpDir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err = common.DefaultNetwork(
+		common.WithNetBadgerPersistence(tmpDir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkErr(t, err)
 	defer n.Close()
 	defer d.Close()
-	d, err = NewDB(context.Background(), n, id, WithNewRepoPath(tmpDir), WithNewKey(info.Key))
+	d, err = NewDB(context.Background(), store, n, id, WithNewKey(info.Key))
 	checkErr(t, err)
 	if d.name != name {
 		t.Fatalf("expected name %s, got %s", name, d.name)
@@ -172,12 +216,20 @@ func TestWithNewEventCodec(t *testing.T) {
 	checkErr(t, err)
 	defer os.RemoveAll(tmpDir)
 
-	n, err := common.DefaultNetwork(tmpDir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(tmpDir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
+	checkErr(t, err)
+
+	store, err := util.NewBadgerDatastore(tmpDir, false)
 	checkErr(t, err)
+	defer store.Close()
 
 	ec := &mockEventCodec{}
 	id := thread.NewIDV1(thread.Raw, 32)
-	d, err := NewDB(context.Background(), n, id, WithNewRepoPath(tmpDir), WithNewEventCodec(ec))
+	d, err := NewDB(context.Background(), store, n, id, WithNewEventCodec(ec))
 	checkErr(t, err)
 
 	m, err := d.NewCollection(CollectionConfig{
@@ -200,10 +252,14 @@ func TestWithNewEventCodec(t *testing.T) {
 	checkErr(t, d.Close())
 
 	time.Sleep(time.Second * 3)
-	n, err = common.DefaultNetwork(tmpDir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err = common.DefaultNetwork(
+		common.WithNetBadgerPersistence(tmpDir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkErr(t, err)
 	defer n.Close()
-	d, err = NewDB(context.Background(), n, id, WithNewRepoPath(tmpDir), WithNewEventCodec(ec), WithNewKey(info.Key))
+	d, err = NewDB(context.Background(), store, n, id, WithNewEventCodec(ec), WithNewKey(info.Key))
 	checkErr(t, err)
 	checkErr(t, d.Close())
 }
@@ -436,7 +492,12 @@ type mockEventCodec struct {
 
 var _ core.EventCodec = (*mockEventCodec)(nil)
 
-func (dec *mockEventCodec) Reduce([]core.Event, ds.TxnDatastore, ds.Key, core.IndexFunc) ([]core.ReduceAction, error) {
+func (dec *mockEventCodec) Reduce(
+	[]core.Event,
+	ds.TxnDatastore,
+	ds.Key,
+	core.IndexFunc,
+) ([]core.ReduceAction, error) {
 	dec.called = true
 	return nil, nil
 }
diff --git a/db/dispatcher_test.go b/db/dispatcher_test.go
index c5b263f..95131ce 100644
--- a/db/dispatcher_test.go
+++ b/db/dispatcher_test.go
@@ -7,8 +7,8 @@ import (
 	"testing"
 	"time"
 
-	datastore "github.com/textileio/go-datastore"
-	"github.com/textileio/go-datastore/query"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	core "github.com/textileio/go-threads/core/db"
 )
 
@@ -96,7 +96,7 @@ func TestValidStore(t *testing.T) {
 	if store == nil {
 		t.Error("store should not be nil")
 	} else {
-		if ok, _ := store.Has(datastore.NewKey("blah")); ok {
+		if ok, _ := store.Has(ds.NewKey("blah")); ok {
 			t.Error("store should be empty")
 		}
 	}
diff --git a/db/manager_test.go b/db/manager_test.go
index e10fc68..ce8bb68 100644
--- a/db/manager_test.go
+++ b/db/manager_test.go
@@ -113,11 +113,18 @@ func TestManager_GetDB(t *testing.T) {
 
 	dir, err := ioutil.TempDir("", "")
 	checkErr(t, err)
-	n, err := common.DefaultNetwork(dir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(dir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkErr(t, err)
-	man, err := NewManager(n, WithNewRepoPath(dir), WithNewDebug(true))
+	store, err := util.NewBadgerDatastore(dir, false)
+	checkErr(t, err)
+	man, err := NewManager(store, n, WithNewDebug(true))
 	checkErr(t, err)
 	defer func() {
+		store.Close()
 		_ = os.RemoveAll(dir)
 	}()
 
@@ -151,9 +158,13 @@ func TestManager_GetDB(t *testing.T) {
 	checkErr(t, err)
 
 	t.Run("test get db after restart", func(t *testing.T) {
-		n, err := common.DefaultNetwork(dir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+		n, err := common.DefaultNetwork(
+			common.WithNetBadgerPersistence(dir),
+			common.WithNetHostAddr(util.FreeLocalAddr()),
+			common.WithNetDebug(true),
+		)
 		checkErr(t, err)
-		man, err := NewManager(n, WithNewRepoPath(dir), WithNewDebug(true))
+		man, err := NewManager(store, n, WithNewDebug(true))
 		checkErr(t, err)
 
 		db, err := man.GetDB(ctx, id)
@@ -184,9 +195,13 @@ func TestManager_GetDB(t *testing.T) {
 		checkErr(t, err)
 
 		t.Run("test get deleted db after restart", func(t *testing.T) {
-			n, err := common.DefaultNetwork(dir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+			n, err := common.DefaultNetwork(
+				common.WithNetBadgerPersistence(dir),
+				common.WithNetHostAddr(util.FreeLocalAddr()),
+				common.WithNetDebug(true),
+			)
 			checkErr(t, err)
-			man, err := NewManager(n, WithNewRepoPath(dir), WithNewDebug(true))
+			man, err := NewManager(store, n, WithNewDebug(true))
 			checkErr(t, err)
 
 			_, err = man.GetDB(ctx, id)
@@ -235,9 +250,15 @@ func TestManager_DeleteDB(t *testing.T) {
 func createTestManager(t *testing.T) (*Manager, func()) {
 	dir, err := ioutil.TempDir("", "")
 	checkErr(t, err)
-	n, err := common.DefaultNetwork(dir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(dir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
+	checkErr(t, err)
+	store, err := util.NewBadgerDatastore(dir, false)
 	checkErr(t, err)
-	m, err := NewManager(n, WithNewRepoPath(dir), WithNewDebug(true))
+	m, err := NewManager(store, n, WithNewDebug(true))
 	checkErr(t, err)
 	return m, func() {
 		if err := n.Close(); err != nil {
@@ -246,6 +267,9 @@ func createTestManager(t *testing.T) (*Manager, func()) {
 		if err := m.Close(); err != nil {
 			panic(err)
 		}
+		if err := store.Close(); err != nil {
+			panic(err)
+		}
 		_ = os.RemoveAll(dir)
 	}
 }
diff --git a/db/testutils_test.go b/db/testutils_test.go
index 3c5d2e7..de5c1ff 100644
--- a/db/testutils_test.go
+++ b/db/testutils_test.go
@@ -22,16 +22,24 @@ func checkErr(t *testing.T, err error) {
 func createTestDB(t *testing.T, opts ...NewOption) (*DB, func()) {
 	dir, err := ioutil.TempDir("", "")
 	checkErr(t, err)
-	n, err := common.DefaultNetwork(dir, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	n, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(dir),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	checkErr(t, err)
-	opts = append(opts, WithNewRepoPath(dir))
-	d, err := NewDB(context.Background(), n, thread.NewIDV1(thread.Raw, 32), opts...)
+	store, err := util.NewBadgerDatastore(dir, false)
+	checkErr(t, err)
+	d, err := NewDB(context.Background(), store, n, thread.NewIDV1(thread.Raw, 32), opts...)
 	checkErr(t, err)
 	return d, func() {
 		time.Sleep(time.Second) // Give threads a chance to finish work
 		if err := n.Close(); err != nil {
 			panic(err)
 		}
+		if err := store.Close(); err != nil {
+			panic(err)
+		}
 		_ = os.RemoveAll(dir)
 	}
 }
diff --git a/integrationtests/foldersync/client.go b/integrationtests/foldersync/client.go
index aa97afe..7c1a19d 100644
--- a/integrationtests/foldersync/client.go
+++ b/integrationtests/foldersync/client.go
@@ -22,6 +22,7 @@ import (
 	"github.com/textileio/go-threads/core/net"
 	"github.com/textileio/go-threads/core/thread"
 	"github.com/textileio/go-threads/db"
+	kt "github.com/textileio/go-threads/db/keytransform"
 	"github.com/textileio/go-threads/integrationtests/foldersync/watcher"
 	"github.com/textileio/go-threads/util"
 )
@@ -62,6 +63,7 @@ type client struct {
 	name           string
 	folderPath     string
 	folderInstance *folder
+	store          kt.TxnDatastoreExtended
 	db             *db.DB
 	collection     *db.Collection
 	net            net.Net
@@ -77,7 +79,11 @@ func newRootClient(name, folderPath, repoPath string) (*client, error) {
 		return nil, err
 	}
 
-	d, err := db.NewDB(context.Background(), network, id, db.WithNewRepoPath(repoPath), db.WithNewCollections(cc))
+	s, err := util.NewBadgerDatastore(repoPath, false)
+	if err != nil {
+		return nil, err
+	}
+	d, err := db.NewDB(context.Background(), s, network, id, db.WithNewCollections(cc))
 	if err != nil {
 		return nil, err
 	}
@@ -86,6 +92,7 @@ func newRootClient(name, folderPath, repoPath string) (*client, error) {
 		name:       name,
 		folderPath: folderPath,
 		db:         d,
+		store:      s,
 		collection: d.GetCollection(collectionName),
 		net:        network,
 		peer:       network.GetIpfsLite(),
@@ -99,7 +106,11 @@ func newJoinerClient(name, folderPath, repoPath string, addr ma.Multiaddr, key t
 		return nil, err
 	}
 
-	d, err := db.NewDBFromAddr(context.Background(), network, addr, key, db.WithNewRepoPath(repoPath), db.WithNewCollections(cc))
+	s, err := util.NewBadgerDatastore(repoPath, false)
+	if err != nil {
+		return nil, err
+	}
+	d, err := db.NewDBFromAddr(context.Background(), s, network, addr, key, db.WithNewCollections(cc))
 	if err != nil {
 		return nil, err
 	}
@@ -108,6 +119,7 @@ func newJoinerClient(name, folderPath, repoPath string, addr ma.Multiaddr, key t
 		name:       name,
 		folderPath: folderPath,
 		db:         d,
+		store:      s,
 		collection: d.GetCollection(collectionName),
 		net:        network,
 		peer:       network.GetIpfsLite(),
@@ -116,7 +128,11 @@ func newJoinerClient(name, folderPath, repoPath string, addr ma.Multiaddr, key t
 }
 
 func newNetwork(repoPath string) (common.NetBoostrapper, error) {
-	network, err := common.DefaultNetwork(repoPath, common.WithNetDebug(true), common.WithNetHostAddr(util.FreeLocalAddr()))
+	network, err := common.DefaultNetwork(
+		common.WithNetBadgerPersistence(repoPath),
+		common.WithNetHostAddr(util.FreeLocalAddr()),
+		common.WithNetDebug(true),
+	)
 	if err != nil {
 		return nil, err
 	}
@@ -332,6 +348,9 @@ func (c *client) close() error {
 	if err := c.db.Close(); err != nil {
 		return err
 	}
+	if err := c.store.Close(); err != nil {
+		return err
+	}
 	if err := c.net.Close(); err != nil {
 		return err
 	}
diff --git a/integrationtests/foldersync/foldersync_test.go b/integrationtests/foldersync/foldersync_test.go
index a078136..115fef8 100644
--- a/integrationtests/foldersync/foldersync_test.go
+++ b/integrationtests/foldersync/foldersync_test.go
@@ -43,7 +43,10 @@ func TestSimple(t *testing.T) {
 	network0, err := newNetwork(repoPath0)
 	checkErr(t, err)
 
-	db0, err := db.NewDB(context.Background(), network0, id, db.WithNewRepoPath(repoPath0), db.WithNewCollections(cc))
+	store0, err := util.NewBadgerDatastore(repoPath0, false)
+	checkErr(t, err)
+	defer store0.Close()
+	db0, err := db.NewDB(context.Background(), store0, network0, id, db.WithNewCollections(cc))
 	checkErr(t, err)
 	defer db0.Close()
 
@@ -60,7 +63,17 @@ func TestSimple(t *testing.T) {
 	network1, err := newNetwork(repoPath1)
 	checkErr(t, err)
 
-	db1, err := db.NewDBFromAddr(context.Background(), network1, info0.Addrs[0], info0.Key, db.WithNewRepoPath(repoPath1), db.WithNewCollections(cc))
+	store1, err := util.NewBadgerDatastore(repoPath1, false)
+	checkErr(t, err)
+	defer store1.Close()
+	db1, err := db.NewDBFromAddr(
+		context.Background(),
+		store1,
+		network1,
+		info0.Addrs[0],
+		info0.Key,
+		db.WithNewCollections(cc),
+	)
 	checkErr(t, err)
 	defer db1.Close()
 
@@ -74,7 +87,17 @@ func TestSimple(t *testing.T) {
 	network2, err := newNetwork(repoPath2)
 	checkErr(t, err)
 
-	db2, err := db.NewDBFromAddr(context.Background(), network2, info0.Addrs[0], info0.Key, db.WithNewRepoPath(repoPath2), db.WithNewCollections(cc))
+	store2, err := util.NewBadgerDatastore(repoPath2, false)
+	checkErr(t, err)
+	defer store2.Close()
+	db2, err := db.NewDBFromAddr(
+		context.Background(),
+		store2,
+		network2,
+		info0.Addrs[0],
+		info0.Key,
+		db.WithNewCollections(cc),
+	)
 	checkErr(t, err)
 	defer db2.Close()
 
@@ -88,7 +111,17 @@ func TestSimple(t *testing.T) {
 	network3, err := newNetwork(repoPath3)
 	checkErr(t, err)
 
-	db3, err := db.NewDBFromAddr(context.Background(), network3, info0.Addrs[0], info0.Key, db.WithNewRepoPath(repoPath3), db.WithNewCollections(cc))
+	store3, err := util.NewBadgerDatastore(repoPath3, false)
+	checkErr(t, err)
+	defer store3.Close()
+	db3, err := db.NewDBFromAddr(
+		context.Background(),
+		store3,
+		network3,
+		info0.Addrs[0],
+		info0.Key,
+		db.WithNewCollections(cc),
+	)
 	checkErr(t, err)
 	defer db3.Close()
 
diff --git a/logstore/lstoreds/ds_test.go b/logstore/lstoreds/ds_test.go
index 88093ef..2a5002c 100644
--- a/logstore/lstoreds/ds_test.go
+++ b/logstore/lstoreds/ds_test.go
@@ -8,7 +8,7 @@ import (
 	"time"
 
 	ds "github.com/ipfs/go-datastore"
-	badger "github.com/ipfs/go-ds-badger"
+	badger "github.com/textileio/go-ds-badger"
 	core "github.com/textileio/go-threads/core/logstore"
 	pt "github.com/textileio/go-threads/test"
 )
diff --git a/logstore/lstorehybrid/hybrid_test.go b/logstore/lstorehybrid/hybrid_test.go
index 0f45018..c5b4375 100644
--- a/logstore/lstorehybrid/hybrid_test.go
+++ b/logstore/lstorehybrid/hybrid_test.go
@@ -6,7 +6,7 @@ import (
 	"os"
 	"testing"
 
-	badger "github.com/ipfs/go-ds-badger"
+	badger "github.com/textileio/go-ds-badger"
 	core "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/logstore/lstoreds"
 	m "github.com/textileio/go-threads/logstore/lstoremem"
diff --git a/net/api/client/client_test.go b/net/api/client/client_test.go
index b0eb949..7d1caf5 100644
--- a/net/api/client/client_test.go
+++ b/net/api/client/client_test.go
@@ -419,14 +419,14 @@ func makeServer(t *testing.T) (ma.Multiaddr, ma.Multiaddr, func()) {
 	}
 	hostAddr := util.FreeLocalAddr()
 	n, err := common.DefaultNetwork(
-		dir,
+		common.WithNetBadgerPersistence(dir),
 		common.WithNetHostAddr(hostAddr),
 		common.WithNetPubSub(true),
-		common.WithNetDebug(true))
+		common.WithNetDebug(true),
+	)
 	if err != nil {
 		t.Fatal(err)
 	}
-	n.Bootstrap(util.DefaultBoostrapPeers())
 	service, err := api.NewService(n, api.Config{
 		Debug: true,
 	})