feat: Add P2P collection topic subscription (#1086)

Relevant issue(s)
Resolves #1081

Description
This PR adds the option to subscribe to events on a given collection by using the collectionID (in reality schemaID for now as there is no globalSchemaID or anything similar) as the pubsub topic. It adds commands to the CLI to add, remove and list the P2P collection topics.

README.md

@@ -314,6 +314,17 @@ About the flags:
 This starts two nodes and connect them via pubsub networking.
 
 
+### Collection subscription example
+
+It is possible to subscribe to updates on a given collection by using its ID as the pubsub topic. After setting up 2 nodes as shown in the [Pubsub example](#pubsub-example) section, we can subscribe to collections update on *nodeA* from *nodeB* by using the `rpc p2pcollection` command:
+```shell
+defradb client rpc p2pcollection add --url localhost:9182 <collectionID>
+```
+Multiple collection IDs can be added at once.
+```shell
+defradb client rpc p2pcollection add --url localhost:9182 <collection1ID> <collection2ID> <collection3ID>
+```
+
 ### Replicator example
 
 Replicator peering is established in one direction. For example, a *nodeA* can be given a *nodeB* to actively send updates to, but *nodeB* won't send updates in return. However, nodes broadcast updates of documents over document-specific pubsub topics, therefore *nodeB* while it won't replicate directly to *nodeA*, it will *passively*.

cli/p2p_collection.go

@@ -0,0 +1,25 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// 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 cli
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var p2pCollectionCmd = &cobra.Command{
+	Use:   "p2pcollection",
+	Short: "Interact with the P2P collection system",
+	Long:  "Add, delete, or get the list of P2P collections",
+}
+
+func init() {
+	rpcCmd.AddCommand(p2pCollectionCmd)
+}

cli/p2p_collection_add.go

@@ -0,0 +1,61 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// 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 cli
+
+import (
+	"context"
+
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/sourcenetwork/defradb/errors"
+	"github.com/sourcenetwork/defradb/logging"
+	netclient "github.com/sourcenetwork/defradb/net/api/client"
+)
+
+var addP2PCollectionCmd = &cobra.Command{
+	Use:   "add [collectionID]",
+	Short: "Add P2P collections",
+	Long:  `Use this command if you wish to add new P2P collections to the pubsub topics`,
+	Args: func(cmd *cobra.Command, args []string) error {
+		if err := cobra.MinimumNArgs(1)(cmd, args); err != nil {
+			return errors.New("must specify at least one collectionID")
+		}
+		return nil
+	},
+	RunE: func(cmd *cobra.Command, args []string) error {
+		cred := insecure.NewCredentials()
+		client, err := netclient.NewClient(cfg.Net.RPCAddress, grpc.WithTransportCredentials(cred))
+		if err != nil {
+			return errors.Wrap("failed to create RPC client", err)
+		}
+
+		rpcTimeoutDuration, err := cfg.Net.RPCTimeoutDuration()
+		if err != nil {
+			return errors.Wrap("failed to parse RPC timeout duration", err)
+		}
+
+		ctx, cancel := context.WithTimeout(cmd.Context(), rpcTimeoutDuration)
+		defer cancel()
+
+		err = client.AddP2PCollections(ctx, args...)
+		if err != nil {
+			return errors.Wrap("failed to add p2p collections, request failed", err)
+		}
+		log.FeedbackInfo(ctx, "Successfully added p2p collections", logging.NewKV("Collections", args))
+		return nil
+	},
+}
+
+func init() {
+	replicatorCmd.AddCommand(addP2PCollectionCmd)
+}

cli/p2p_collection_getall.go

@@ -0,0 +1,70 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// 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 cli
+
+import (
+	"context"
+
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/sourcenetwork/defradb/errors"
+	"github.com/sourcenetwork/defradb/logging"
+	netclient "github.com/sourcenetwork/defradb/net/api/client"
+)
+
+var getAllP2PCollectionCmd = &cobra.Command{
+	Use:   "getall",
+	Short: "Get all P2P collections",
+	Long:  `Use this command if you wish to get all P2P collections in the pubsub topics`,
+	Args: func(cmd *cobra.Command, args []string) error {
+		if err := cobra.NoArgs(cmd, args); err != nil {
+			return errors.New("must specify no argument")
+		}
+		return nil
+	},
+	RunE: func(cmd *cobra.Command, args []string) error {
+		cred := insecure.NewCredentials()
+		client, err := netclient.NewClient(cfg.Net.RPCAddress, grpc.WithTransportCredentials(cred))
+		if err != nil {
+			return errors.Wrap("failed to create RPC client", err)
+		}
+
+		rpcTimeoutDuration, err := cfg.Net.RPCTimeoutDuration()
+		if err != nil {
+			return errors.Wrap("failed to parse RPC timeout duration", err)
+		}
+
+		ctx, cancel := context.WithTimeout(cmd.Context(), rpcTimeoutDuration)
+		defer cancel()
+
+		collections, err := client.GetAllP2PCollections(ctx)
+		if err != nil {
+			return errors.Wrap("failed to add p2p collections, request failed", err)
+		}
+
+		if len(collections) > 0 {
+			log.FeedbackInfo(ctx, "Successfully got all P2P collections")
+			for _, col := range collections {
+				log.FeedbackInfo(ctx, col.Name, logging.NewKV("CollectionID", col.ID))
+			}
+		} else {
+			log.FeedbackInfo(ctx, "No P2P collection found")
+		}
+
+		return nil
+	},
+}
+
+func init() {
+	replicatorCmd.AddCommand(getAllP2PCollectionCmd)
+}

cli/p2p_collection_remove.go

@@ -0,0 +1,61 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// 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 cli
+
+import (
+	"context"
+
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/sourcenetwork/defradb/errors"
+	"github.com/sourcenetwork/defradb/logging"
+	netclient "github.com/sourcenetwork/defradb/net/api/client"
+)
+
+var removeP2PCollectionCmd = &cobra.Command{
+	Use:   "remove [collectionID]",
+	Short: "Add P2P collections",
+	Long:  `Use this command if you wish to remove P2P collections from the pubsub topics`,
+	Args: func(cmd *cobra.Command, args []string) error {
+		if err := cobra.MinimumNArgs(1)(cmd, args); err != nil {
+			return errors.New("must specify at least one collectionID")
+		}
+		return nil
+	},
+	RunE: func(cmd *cobra.Command, args []string) error {
+		cred := insecure.NewCredentials()
+		client, err := netclient.NewClient(cfg.Net.RPCAddress, grpc.WithTransportCredentials(cred))
+		if err != nil {
+			return errors.Wrap("failed to create RPC client", err)
+		}
+
+		rpcTimeoutDuration, err := cfg.Net.RPCTimeoutDuration()
+		if err != nil {
+			return errors.Wrap("failed to parse RPC timeout duration", err)
+		}
+
+		ctx, cancel := context.WithTimeout(cmd.Context(), rpcTimeoutDuration)
+		defer cancel()
+
+		err = client.RemoveP2PCollections(ctx, args...)
+		if err != nil {
+			return errors.Wrap("failed to remove p2p collections, request failed", err)
+		}
+		log.FeedbackInfo(ctx, "Successfully removed p2p collections", logging.NewKV("Collections", args))
+		return nil
+	},
+}
+
+func init() {
+	replicatorCmd.AddCommand(removeP2PCollectionCmd)
+}

client/collection.go

@@ -166,3 +166,11 @@ type DeleteResult struct {
 	// DocKeys contains the DocKeys of all the documents deleted by the delete call.
 	DocKeys []string
 }
+
+// P2PCollection is the gRPC response representation of a P2P collection topic
+type P2PCollection struct {
+	// The collection ID
+	ID string
+	// The Collection name
+	Name string
+}

client/db.go

@@ -42,15 +42,8 @@ type DB interface {
 
 	PrintDump(ctx context.Context) error
 
-	// SetReplicator adds a replicator to the persisted list or adds
-	// schemas if the replicator already exists.
-	SetReplicator(ctx context.Context, rep Replicator) error
-	// DeleteReplicator deletes a replicator from the persisted list
-	// or specific schemas if they are specified.
-	DeleteReplicator(ctx context.Context, rep Replicator) error
-	// GetAllReplicators returns the full list of replicators with their
-	// subscribed schemas.
-	GetAllReplicators(ctx context.Context) ([]Replicator, error)
+	// P2P holds the P2P related methods that must be implemented by the database.
+	P2P
 }
 
 type GQLResult struct {

client/p2p.go

@@ -0,0 +1,37 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// 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 client
+
+import "context"
+
+type P2P interface {
+	// SetReplicator adds a replicator to the persisted list or adds
+	// schemas if the replicator already exists.
+	SetReplicator(ctx context.Context, rep Replicator) error
+	// DeleteReplicator deletes a replicator from the persisted list
+	// or specific schemas if they are specified.
+	DeleteReplicator(ctx context.Context, rep Replicator) error
+	// GetAllReplicators returns the full list of replicators with their
+	// subscribed schemas.
+	GetAllReplicators(ctx context.Context) ([]Replicator, error)
+
+	// AddP2PCollection adds the given collection ID that the P2P system
+	// subscribes to to the the persisted list. It will error if the provided
+	// collection ID is invalid.
+	AddP2PCollection(ctx context.Context, collectionID string) error
+	// RemoveP2PCollection removes the given collection ID that the P2P system
+	// subscribes to from the the persisted list. It will error if the provided
+	// collection ID is invalid.
+	RemoveP2PCollection(ctx context.Context, collectionID string) error
+	// GetAllP2PCollections returns the list of persisted collection IDs that
+	// the P2P system subscribes to.
+	GetAllP2PCollections(ctx context.Context) ([]string, error)
+}

core/key.go

@@ -18,6 +18,7 @@ import (
 	ds "github.com/ipfs/go-datastore"
 
 	"github.com/sourcenetwork/defradb/client"
+	"github.com/sourcenetwork/defradb/errors"
 )
 
 var (
@@ -45,6 +46,7 @@ const (
 	SEQ                       = "/seq"
 	PRIMARY_KEY               = "/pk"
 	REPLICATOR                = "/replicator/id"
+	P2P_COLLECTION            = "/p2p/collection"
 )
 
 // Key is an interface that represents a key in the database.
@@ -103,6 +105,12 @@ type CollectionSchemaVersionKey struct {
 
 var _ Key = (*CollectionSchemaVersionKey)(nil)
 
+type P2PCollectionKey struct {
+	CollectionID string
+}
+
+var _ Key = (*P2PCollectionKey)(nil)
+
 type SchemaKey struct {
 	SchemaName string
 }
@@ -434,6 +442,37 @@ func (k SequenceKey) ToDS() ds.Key {
 	return ds.NewKey(k.ToString())
 }
 
+// New
+func NewP2PCollectionKey(collectionID string) P2PCollectionKey {
+	return P2PCollectionKey{CollectionID: collectionID}
+}
+
+func NewP2PCollectionKeyFromString(key string) (P2PCollectionKey, error) {
+	keyArr := strings.Split(key, "/")
+	if len(keyArr) != 4 {
+		return P2PCollectionKey{}, errors.WithStack(ErrInvalidKey, errors.NewKV("Key", key))
+	}
+	return NewP2PCollectionKey(keyArr[3]), nil
+}
+
+func (k P2PCollectionKey) ToString() string {
+	result := P2P_COLLECTION
+
+	if k.CollectionID != "" {
+		result = result + "/" + k.CollectionID
+	}
+
+	return result
+}
+
+func (k P2PCollectionKey) Bytes() []byte {
+	return []byte(k.ToString())
+}
+
+func (k P2PCollectionKey) ToDS() ds.Key {
+	return ds.NewKey(k.ToString())
+}
+
 func NewReplicatorKey(id string) ReplicatorKey {
 	return ReplicatorKey{ReplicatorID: id}
 }