refactor: P2P client interface

README.md

@@ -244,57 +244,73 @@ When starting a node for the first time, a key pair is generated and stored in i
 
 Each node has a unique `PeerID` generated from its public key. This ID allows other nodes to connect to it.
 
+To view your node's peer info:
+
+```shell
+defradb client p2p info
+```
+
 There are two types of peer-to-peer relationships supported: **pubsub** peering and **replicator** peering.
 
 Pubsub peering *passively* synchronizes data between nodes by broadcasting *Document Commit* updates to the topic of the commit's document key. Nodes need to be listening on the pubsub channel to receive updates. This is for when two nodes *already* have share a document and want to keep them in sync.
 
 Replicator peering *actively* pushes changes from a specific collection *to* a target peer.
 
-### Pubsub example
+<details>
+<summary>Pubsub example</summary>
 
 Pubsub peers can be specified on the command line using the `--peers` flag, which accepts a comma-separated list of peer [multiaddresses](https://docs.libp2p.io/concepts/addressing/). For example, a node at IP `192.168.1.12` listening on 9000 with PeerID `12D3KooWNXm3dmrwCYSxGoRUyZstaKYiHPdt8uZH5vgVaEJyzU8B` would be referred to using the multiaddress `/ip4/192.168.1.12/tcp/9000/p2p/12D3KooWNXm3dmrwCYSxGoRUyZstaKYiHPdt8uZH5vgVaEJyzU8B`.
 
 Let's go through an example of two nodes (*nodeA* and *nodeB*) connecting with each other over pubsub, on the same machine.
 
 Start *nodeA* with a default configuration:
 
-```
+```shell
 defradb start
 ```
 
-Obtain the PeerID from its console output. In this example, we use `12D3KooWNXm3dmrwCYSxGoRUyZstaKYiHPdt8uZH5vgVaEJyzU8B`, but locally it will be different.
+Obtain the node's peer info:
+
+```shell
+defradb client p2p info
+```
+
+In this example, we use `12D3KooWNXm3dmrwCYSxGoRUyZstaKYiHPdt8uZH5vgVaEJyzU8B`, but locally it will be different.
 
 For *nodeB*, we provide the following configuration:
 
-```
-defradb start --rootdir ~/.defradb-nodeB --url localhost:9182 --p2paddr /ip4/0.0.0.0/tcp/9172 --tcpaddr /ip4/0.0.0.0/tcp/9162 --peers /ip4/0.0.0.0/tcp/9171/p2p/12D3KooWNXm3dmrwCYSxGoRUyZstaKYiHPdt8uZH5vgVaEJyzU8B
+```shell
+defradb start --rootdir ~/.defradb-nodeB --url localhost:9182 --p2paddr /ip4/0.0.0.0/tcp/9172 --peers /ip4/0.0.0.0/tcp/9171/p2p/12D3KooWNXm3dmrwCYSxGoRUyZstaKYiHPdt8uZH5vgVaEJyzU8B
 ```
 
 About the flags:
 
 - `--rootdir` specifies the root dir (config and data) to use
 - `--url` is the address to listen on for the client HTTP and GraphQL API
 - `--p2paddr` is the multiaddress for the P2P networking to listen on
-- `--tcpaddr` is the multiaddress for the gRPC server to listen on
 - `--peers` is a comma-separated list of peer multiaddresses
 
 This starts two nodes and connects them via pubsub networking.
+</details>
 
-### Collection subscription example
+<details>
+<summary>Subscription example</summary>
 
-It is possible to subscribe to updates on a given collection by using its ID as the pubsub topic. The ID of a collection is found as the field `collectionID` in one of its documents. Here we use the collection ID of the `User` type we created above. After setting up 2 nodes as shown in the [Pubsub example](#pubsub-example) section, we can subscribe to collections updates on *nodeA* from *nodeB* by using the `rpc p2pcollection` command:
+It is possible to subscribe to updates on a given collection by using its ID as the pubsub topic. The ID of a collection is found as the field `collectionID` in one of its documents. Here we use the collection ID of the `User` type we created above. After setting up 2 nodes as shown in the [Pubsub example](#pubsub-example) section, we can subscribe to collections updates on *nodeA* from *nodeB* by using the following command:
 
 ```shell
-defradb client rpc p2pcollection add --url localhost:9182 bafkreibpnvkvjqvg4skzlijka5xe63zeu74ivcjwd76q7yi65jdhwqhske
+defradb client p2p collection add --url localhost:9182 bafkreibpnvkvjqvg4skzlijka5xe63zeu74ivcjwd76q7yi65jdhwqhske
 ```
 
 Multiple collection IDs can be added at once.
 
 ```shell
-defradb client rpc p2pcollection add --url localhost:9182 <collection1ID> <collection2ID> <collection3ID>
+defradb client p2p collection add --url localhost:9182 <collection1ID>,<collection2ID>,<collection3ID>
 ```
+</details>
 
-### Replicator example
+<details>
+<summary>Replicator example</summary>
 
 Replicator peering is targeted: it allows a node to actively send updates to another node. Let's go through an example of *nodeA* actively replicating to *nodeB*:
 
@@ -334,14 +350,20 @@ defradb client schema add --url localhost:9182 '
 '
 ```
 
-Set *nodeA* to actively replicate the "Article" collection to *nodeB*:
+Then copy the peer info from *nodeB*:
 
 ```shell
-defradb client rpc replicator set -c "Article" /ip4/0.0.0.0/tcp/9172/p2p/<peerID_of_nodeB>
+defradb client p2p info
 ```
 
-As we add or update documents in the "Article" collection on *nodeA*, they will be actively pushed to *nodeB*. Note that changes to *nodeB* will still be passively published back to *nodeA*, via pubsub.
+Set *nodeA* to actively replicate the Article collection to *nodeB*:
+
+```shell
+defradb client p2p replicator set -c Article <peer_info_of_nodeB>
+```
 
+As we add or update documents in the Article collection on *nodeA*, they will be actively pushed to *nodeB*. Note that changes to *nodeB* will still be passively published back to *nodeA*, via pubsub.
+</details>
 
 ## Securing the HTTP API with TLS
 

cli/p2p_collection_add.go

@@ -31,7 +31,7 @@ Example: add multiple collections
 		`,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
 			var collectionIDs []string
 			for _, id := range strings.Split(args[0], ",") {
@@ -42,7 +42,7 @@ Example: add multiple collections
 				collectionIDs = append(collectionIDs, id)
 			}
 
-			return store.AddP2PCollections(cmd.Context(), collectionIDs)
+			return p2p.AddP2PCollections(cmd.Context(), collectionIDs)
 		},
 	}
 	return cmd

cli/p2p_collection_getall.go

@@ -22,9 +22,9 @@ func MakeP2PCollectionGetAllCommand() *cobra.Command {
 This is the list of collections of the node that are synchronized on the pubsub network.`,
 		Args: cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			cols, err := store.GetAllP2PCollections(cmd.Context())
+			cols, err := p2p.GetAllP2PCollections(cmd.Context())
 			if err != nil {
 				return err
 			}

cli/p2p_collection_remove.go

@@ -31,7 +31,7 @@ Example: remove multiple collections
 		`,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
 			var collectionIDs []string
 			for _, id := range strings.Split(args[0], ",") {
@@ -42,7 +42,7 @@ Example: remove multiple collections
 				collectionIDs = append(collectionIDs, id)
 			}
 
-			return store.RemoveP2PCollections(cmd.Context(), collectionIDs)
+			return p2p.RemoveP2PCollections(cmd.Context(), collectionIDs)
 		},
 	}
 	return cmd

cli/p2p_info.go

@@ -23,12 +23,7 @@ func MakeP2PInfoCommand() *cobra.Command {
 		Long:  `Get peer info from a DefraDB node`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			db := cmd.Context().Value(dbContextKey).(*http.Client)
-
-			res, err := db.PeerInfo(cmd.Context())
-			if err != nil {
-				return err
-			}
-			return writeJSON(cmd, res)
+			return writeJSON(cmd, db.PeerInfo())
 		},
 	}
 	return cmd

cli/p2p_replicator_delete.go

@@ -11,27 +11,40 @@
 package cli
 
 import (
+	"encoding/json"
+
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/spf13/cobra"
 
 	"github.com/sourcenetwork/defradb/client"
 )
 
 func MakeP2PReplicatorDeleteCommand() *cobra.Command {
+	var collections []string
 	var cmd = &cobra.Command{
-		Use:   "delete <peer>",
+		Use:   "delete [-c, --collection] <peer>",
 		Short: "Delete a replicator. It will stop synchronizing",
-		Long:  `Delete a replicator. It will stop synchronizing.`,
-		Args:  cobra.ExactArgs(1),
+		Long: `Delete a replicator. It will stop synchronizing.
+
+Example:		
+  defradb client p2p replicator delete -c Users '{"ID": "12D3", "Addrs": ["/ip4/0.0.0.0/tcp/9171"]}'
+		`,
+		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			addr, err := peer.AddrInfoFromString(args[0])
-			if err != nil {
+			var info peer.AddrInfo
+			if err := json.Unmarshal([]byte(args[0]), &info); err != nil {
 				return err
 			}
-			return store.DeleteReplicator(cmd.Context(), client.Replicator{Info: *addr})
+			rep := client.Replicator{
+				Info:    info,
+				Schemas: collections,
+			}
+			return p2p.DeleteReplicator(cmd.Context(), rep)
 		},
 	}
+	cmd.Flags().StringSliceVarP(&collections, "collection", "c",
+		[]string{}, "Collection(s) to stop replicating")
 	return cmd
 }

cli/p2p_replicator_getall.go

@@ -21,9 +21,9 @@
 		Long: `Get all the replicators active in the P2P data sync system.
 These are the replicators that are currently replicating data from one node to another.`,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			reps, err := store.GetAllReplicators(cmd.Context())
+			reps, err := p2p.GetAllReplicators(cmd.Context())
 			if err != nil {
 				return err
 			}

cli/p2p_replicator_set.go

@@ -11,6 +11,8 @@
 package cli
 
 import (
+	"encoding/json"
+
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/spf13/cobra"
 
@@ -24,24 +26,27 @@ func MakeP2PReplicatorSetCommand() *cobra.Command {
 		Short: "Set a P2P replicator",
 		Long: `Add a new target replicator.
 A replicator replicates one or all collection(s) from this node to another.
+
+Example:
+  defradb client p2p replicator set -c Users '{"ID": "12D3", "Addrs": ["/ip4/0.0.0.0/tcp/9171"]}'
 `,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			addr, err := peer.AddrInfoFromString(args[0])
-			if err != nil {
+			var info peer.AddrInfo
+			if err := json.Unmarshal([]byte(args[0]), &info); err != nil {
 				return err
 			}
 			rep := client.Replicator{
-				Info:    *addr,
+				Info:    info,
 				Schemas: collections,
 			}
-			return store.SetReplicator(cmd.Context(), rep)
+			return p2p.SetReplicator(cmd.Context(), rep)
 		},
 	}
 
 	cmd.Flags().StringSliceVarP(&collections, "collection", "c",
-		[]string{}, "Define the collection for the replicator")
+		[]string{}, "Collection(s) to replicate")
 	return cmd
 }

cli/start.go

@@ -171,15 +171,10 @@
 
 func (di *defraInstance) close(ctx context.Context) {
 	if di.node != nil {
-		if err := di.node.Close(); err != nil {
-			log.FeedbackInfo(
-				ctx,
-				"The node could not be closed successfully",
-				logging.NewKV("Error", err.Error()),
-			)
-		}
+		di.node.Close()
+	} else {
+		di.db.Close()
 	}
-	di.db.Close(ctx)
 	if err := di.server.Close(); err != nil {
 		log.FeedbackInfo(
 			ctx,
@@ -222,7 +217,7 @@
 	}
 
 	// init the p2p node
-	var n *net.Node
+	var node *net.Node
 	if !cfg.Net.P2PDisabled {
 		nodeOpts := []net.NodeOpt{
 			net.WithConfig(cfg),
@@ -239,9 +234,9 @@
 			nodeOpts = append(nodeOpts, net.WithPrivateKey(key))
 		}
 		log.FeedbackInfo(ctx, "Starting P2P node", logging.NewKV("P2P address", cfg.Net.P2PAddress))
-		n, err = net.NewNode(ctx, db, nodeOpts...)
+		node, err = net.NewNode(ctx, db, nodeOpts...)
 		if err != nil {
-			db.Close(ctx)
+			db.Close()
 			return nil, errors.Wrap("failed to start P2P node", err)
 		}
 
@@ -253,14 +248,11 @@
 				return nil, errors.Wrap(fmt.Sprintf("failed to parse bootstrap peers %v", cfg.Net.Peers), err)
 			}
 			log.Debug(ctx, "Bootstrapping with peers", logging.NewKV("Addresses", addrs))
-			n.Bootstrap(addrs)
+			node.Bootstrap(addrs)
 		}
 
-		if err := n.Start(); err != nil {
-			if e := n.Close(); e != nil {
-				err = errors.Wrap(fmt.Sprintf("failed to close node: %v", e.Error()), err)
-			}
-			db.Close(ctx)
+		if err := node.Start(); err != nil {
+			node.Close()
 			return nil, errors.Wrap("failed to start P2P listeners", err)
 		}
 	}
@@ -271,10 +263,6 @@
 		httpapi.WithAllowedOrigins(cfg.API.AllowedOrigins...),
 	}
 
-	if n != nil {
-		sOpt = append(sOpt, httpapi.WithPeerID(n.PeerID().String()))
-	}
-
 	if cfg.API.TLS {
 		sOpt = append(
 			sOpt,
@@ -284,32 +272,36 @@
 		)
 	}
 
-	s := httpapi.NewServer(db, sOpt...)
-	if err := s.Listen(ctx); err != nil {
-		return nil, errors.Wrap(fmt.Sprintf("failed to listen on TCP address %v", s.Addr), err)
+	var server *httpapi.Server
+	if node != nil {
+		server = httpapi.NewServer(node, sOpt...)
+	} else {
+		server = httpapi.NewServer(db, sOpt...)
+	}
+	if err := server.Listen(ctx); err != nil {
+		return nil, errors.Wrap(fmt.Sprintf("failed to listen on TCP address %v", server.Addr), err)
 	}
 	// save the address on the config in case the port number was set to random
-	cfg.API.Address = s.AssignedAddr()
+	cfg.API.Address = server.AssignedAddr()
 
 	// run the server in a separate goroutine
 	go func() {
 		log.FeedbackInfo(ctx, fmt.Sprintf("Providing HTTP API at %s.", cfg.API.AddressToURL()))
-		if err := s.Run(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		if err := server.Run(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			log.FeedbackErrorE(ctx, "Failed to run the HTTP server", err)
-			if n != nil {
-				if err := n.Close(); err != nil {
-					log.FeedbackErrorE(ctx, "Failed to close node", err)
-				}
+			if node != nil {
+				node.Close()
+			} else {
+				db.Close()
 			}
-			db.Close(ctx)
 			os.Exit(1)
 		}
 	}()
 
 	return &defraInstance{
-		node:   n,
+		node:   node,
 		db:     db,
-		server: s,
+		server: server,
 	}, nil
 }