feat: Added Peer-to-Peer networking data synchronization

api/http/api.go

@@ -55,10 +55,12 @@ func NewServer(db client.DB) *Server {
 	return s
 }
 
-func (s *Server) Listen(addr string) {
+func (s *Server) Listen(addr string) error {
 	if err := http.ListenAndServe(addr, s.router); err != nil {
 		log.Fatalln("Error: HTTP Listening and Serving Failed: ", err)
+		return err
 	}
+	return nil
 }
 
 func (s *Server) ping(w http.ResponseWriter, r *http.Request) {

cli/defradb/cmd/config.go

@@ -15,6 +15,7 @@ import (
 
 type Config struct {
 	Database Options
+	Net      NetOptions
 }
 
 type Options struct {
@@ -35,6 +36,11 @@ type MemoryOptions struct {
 	Size uint64
 }
 
+type NetOptions struct {
+	P2PAddress  string
+	P2PDisabled bool
+}
+
 var (
 	defaultConfig = Config{
 		Database: Options{
@@ -44,5 +50,8 @@ var (
 				Path: "$HOME/.defradb/data",
 			},
 		},
+		Net: NetOptions{
+			P2PAddress: "/ip4/0.0.0.0/tcp/9171",
+		},
 	}
 )

cli/defradb/cmd/root.go

@@ -11,6 +11,7 @@ package cmd
 
 import (
 	"bytes"
+	"fmt"
 	"os"
 	"strings"
 
@@ -63,7 +64,7 @@ func init() {
 	// Here you will define your flags and configuration settings.
 	// Cobra supports persistent flags, which, if defined here,
 	// will be global for your application.
-	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is $HOME/.defradb.yaml)")
+	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is $HOME/.defradb/config.yaml)")
 	rootCmd.PersistentFlags().StringVar(&dbURL, "url", "http://localhost:9181", "url of the target database")
 	// Cobra also supports local flags, which will only run
 	// when this action is called directly.
@@ -74,11 +75,32 @@ func init() {
 }
 
 func initLogger() {
-	lvl, err := logging.LevelFromString(logLvl)
-	if err != nil {
-		panic(err)
+	lvls := strings.Split(logLvl, ",")
+	if len(lvls) == 1 {
+		lvl, err := logging.LevelFromString(logLvl)
+		if err != nil {
+			panic(err)
+		}
+		logging.SetAllLoggers(lvl)
+	} else {
+		lvl, err := logging.LevelFromString(lvls[0])
+		if err != nil {
+			panic(err)
+		}
+		logging.SetAllLoggers(lvl)
+
+		for _, l := range lvls[1:] {
+			lvl := strings.Split(l, "=")
+			if len(lvl) != 2 {
+				fmt.Printf("Invalid format for log level: %s\n", l)
+				os.Exit(1)
+			}
+			if err := logging.SetLogLevel(lvl[0], lvl[1]); err != nil {
+				fmt.Printf("Failed to set log level: %s\n", err)
+				os.Exit(1)
+			}
+		}
 	}
-	logging.SetAllLoggers(lvl)
 }
 
 // initConfig reads in config file and ENV variables if set.
@@ -106,8 +128,11 @@ func initConfig() {
 		// fmt.Fprintln(os.Stderr, "Using config file:", viper.ConfigFileUsed())
 		log.Debug("Loading config file:", viper.ConfigFileUsed())
 	} else {
-		if err := os.Mkdir(home+"/.defradb", os.ModePerm); err != nil {
-			cobra.CheckErr(err)
+		dir := home + "/.defradb"
+		if _, err := os.Stat(dir); os.IsNotExist(err) {
+			if err := os.Mkdir(dir, os.ModePerm); err != nil {
+				cobra.CheckErr(err)
+			}
 		}
 		// if err != nil {
 		// 	cobra.CheckErr(err)
@@ -130,6 +155,15 @@ func initConfig() {
 	err = viper.BindPFlag("database.store", startCmd.Flags().Lookup("store"))
 	cobra.CheckErr(err)
 
+	err = viper.BindPFlag("database.badger.path", startCmd.Flags().Lookup("data"))
+	cobra.CheckErr(err)
+
+	err = viper.BindPFlag("net.p2paddress", startCmd.Flags().Lookup("p2paddr"))
+	cobra.CheckErr(err)
+
+	err = viper.BindPFlag("net.p2pdisabled", startCmd.Flags().Lookup("no-p2p"))
+	cobra.CheckErr(err)
+
 	err = viper.Unmarshal(&config)
 	cobra.CheckErr(err)
 }

cli/defradb/cmd/rpc.go

@@ -0,0 +1 @@
+package cmd

cli/defradb/cmd/serverdump.go

@@ -0,0 +1,69 @@
+// Copyright 2020 Source Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+package cmd
+
+import (
+	"context"
+	"os"
+	"os/signal"
+
+	ds "github.com/ipfs/go-datastore"
+	badgerds "github.com/sourcenetwork/defradb/datastores/badger/v3"
+	"github.com/spf13/cobra"
+
+	"github.com/sourcenetwork/defradb/db"
+)
+
+// dumpCmd represents the dump command
+var srvDumpCmd = &cobra.Command{
+	Use:   "server-dump",
+	Short: "Dumps the state of the entire database (server side)",
+	Run: func(cmd *cobra.Command, args []string) {
+		log.Info("Starting DefraDB process...")
+		ctx := context.Background()
+
+		// setup signal handlers
+		signalCh := make(chan os.Signal, 1)
+		signal.Notify(signalCh, os.Interrupt)
+
+		var rootstore ds.Batching
+		var err error
+		if config.Database.Store == "badger" {
+			log.Info("opening badger store: ", config.Database.Badger.Path)
+			rootstore, err = badgerds.NewDatastore(config.Database.Badger.Path, config.Database.Badger.Options)
+		} else {
+			log.Error("Server side dump is only supported for the Badger datastore")
+			os.Exit(1)
+		}
+		if err != nil {
+			log.Error("Failed to initiate datastore:", err)
+			os.Exit(1)
+		}
+
+		db, err := db.NewDB(rootstore)
+		if err != nil {
+			log.Error("Failed to initiate database:", err)
+			os.Exit(1)
+		}
+		if err := db.Start(ctx); err != nil {
+			log.Error("Failed to start the database: ", err)
+			db.Close()
+			os.Exit(1)
+		}
+
+		log.Info("Dumping DB state:")
+		db.PrintDump(ctx)
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(srvDumpCmd)
+	srvDumpCmd.Flags().String("store", "badger", "Specify the data store to use (supported: badger, memory)")
+}

cli/defradb/cmd/start.go

@@ -13,13 +13,25 @@ import (
 	"context"
 	"os"
 	"os/signal"
+	"strings"
 
+	badgerds "github.com/sourcenetwork/defradb/datastores/badger/v3"
 	"github.com/sourcenetwork/defradb/db"
+	netutils "github.com/sourcenetwork/defradb/net/utils"
+	"github.com/sourcenetwork/defradb/node"
 
 	badger "github.com/dgraph-io/badger/v3"
 	ds "github.com/ipfs/go-datastore"
-	badgerds "github.com/sourcenetwork/defradb/datastores/badger/v3"
 	"github.com/spf13/cobra"
+	"github.com/textileio/go-threads/broadcast"
+)
+
+var (
+	p2pAddr  string
+	dataPath string
+	peers    string
+
+	busBufferSize = 100
 )
 
 // startCmd represents the start command
@@ -36,25 +48,32 @@ var startCmd = &cobra.Command{
 		signal.Notify(signalCh, os.Interrupt)
 
 		var rootstore ds.Batching
-		var options interface{}
+
 		var err error
 		if config.Database.Store == "badger" {
 			log.Info("opening badger store: ", config.Database.Badger.Path)
 			rootstore, err = badgerds.NewDatastore(config.Database.Badger.Path, config.Database.Badger.Options)
-			options = config.Database.Badger
 		} else if config.Database.Store == "memory" {
 			log.Info("building new memory store")
 			opts := badgerds.Options{Options: badger.DefaultOptions("").WithInMemory(true)}
 			rootstore, err = badgerds.NewDatastore("", &opts)
-			options = config.Database.Memory
 		}
 
 		if err != nil {
 			log.Error("Failed to initiate datastore:", err)
 			os.Exit(1)
 		}
 
-		db, err := db.NewDB(rootstore, options)
+		var options []db.Option
+
+		// check for p2p
+		var bs *broadcast.Broadcaster
+		if !config.Net.P2PDisabled {
+			bs = broadcast.NewBroadcaster(busBufferSize)
+			options = append(options, db.WithBroadcaster(bs))
+		}
+
+		db, err := db.NewDB(rootstore, options...)
 		if err != nil {
 			log.Error("Failed to initiate database:", err)
 			os.Exit(1)
@@ -65,13 +84,60 @@ var startCmd = &cobra.Command{
 			os.Exit(1)
 		}
 
-		// run the server listener in a separate goroutine
+		// init the p2p node
+		var n *node.Node
+		if !config.Net.P2PDisabled {
+			n, err = node.NewNode(
+				ctx,
+				db,
+				bs,
+				node.DataPath(config.Database.Badger.Path),
+				node.ListenAddrStrings(config.Net.P2PAddress),
+				node.WithPubSub(true))
+			if err != nil {
+				log.Error("Failed to start p2p node:", err)
+				n.Close() //nolint
+				db.Close()
+				os.Exit(1)
+			}
+
+			// parse peers and bootstrap
+			if len(peers) != 0 {
+				log.Debug("Parsing boostrap peers: ", peers)
+				addrs, err := netutils.ParsePeers(strings.Split(peers, ","))
+				if err != nil {
+					log.Warn("Failed to parse boostrap peers: ", err)
+				}
+				log.Debug("Bootstraping with peers: ", addrs)
+				n.Boostrap(addrs)
+			}
+
+			if err := n.Start(); err != nil {
+				log.Error("Failed to start p2p listener:", err)
+				n.Close() //nolint
+				db.Close()
+				os.Exit(1)
+			}
+		}
+
+		// run the server listener in a seperate goroutine
 		go func() {
-			db.Listen(config.Database.Address)
+			if err := db.Listen(config.Database.Address); err != nil {
+				log.Error("Failed to start API listener:", err)
+				if n != nil {
+					n.Close() //nolint
+				}
+				db.Close()
+				os.Exit(1)
+			}
 		}()
 
+		// wait for shutdown signal
 		<-signalCh
 		log.Info("Recieved interrupt; closing db")
+		if n != nil {
+			n.Close() //nolint
+		}
 		db.Close()
 		os.Exit(0)
 	},
@@ -89,5 +155,8 @@ func init() {
 	// Cobra supports local flags which will only run when this command
 	// is called directly, e.g.:
 	startCmd.Flags().String("store", "badger", "Specify the data store to use (supported: badger, memory)")
-
+	startCmd.Flags().StringVar(&peers, "peers", "", "list of peers to connect to")
+	startCmd.Flags().StringVar(&p2pAddr, "p2paddr", "/ip4/0.0.0.0/tcp/9171", "listener address for the p2p network (formatted as a libp2p MultiAddr)")
+	startCmd.Flags().StringVar(&dataPath, "data", "$HOME/.defradb/data", "Data path to save DB data and other related meta-data")
+	startCmd.Flags().Bool("no-p2p", false, "Turn off the peer-to-peer network synchroniation system")
 }

cli/defradb/examples/address.graphql

@@ -0,0 +1,6 @@
+type address {
+	street: String 
+	number: Int 
+	city: String
+    country: String
+}

client/core.go

@@ -21,17 +21,27 @@ import (
 
 	blocks "github.com/ipfs/go-block-format"
 	cid "github.com/ipfs/go-cid"
+	ds "github.com/ipfs/go-datastore"
 )
 
 type DB interface {
 	// Collections
 	CreateCollection(context.Context, base.CollectionDescription) (Collection, error)
 	GetCollection(context.Context, string) (Collection, error)
+	GetCollectionBySchemaID(context.Context, string) (Collection, error)
 	ExecQuery(context.Context, string) *QueryResult
 	SchemaManager() *schema.SchemaManager
 	AddSchema(context.Context, string) error
 	PrintDump(ctx context.Context)
 	GetBlock(ctx context.Context, c cid.Cid) (blocks.Block, error)
+	Root() ds.Batching
+	// Rootstore() core.DSReaderWriter
+	// Headstore() core.DSReaderWriter
+	// Datastore() core.DSReaderWriter
+	DAGstore() core.DAGStore
+
+	NewTxn(context.Context, bool) (Txn, error)
+	GetAllCollections(ctx context.Context) ([]Collection, error)
 }
 
 type Sequence interface{}
@@ -50,6 +60,7 @@ type Collection interface {
 	Name() string
 	Schema() base.SchemaDescription
 	ID() uint32
+	SchemaID() string
 
 	Indexes() []base.IndexDescription
 	PrimaryIndex() base.IndexDescription
@@ -76,6 +87,14 @@ type Collection interface {
 	Get(context.Context, key.DocKey) (*document.Document, error)
 
 	WithTxn(Txn) Collection
+
+	GetPrimaryIndexDocKey(ds.Key) ds.Key
+	GetAllDocKeys(ctx context.Context) (<-chan DocKeysResult, error)
+}
+
+type DocKeysResult struct {
+	Key key.DocKey
+	Err error
 }
 
 type UpdateOpt struct{}