Add query endpoint for subcribers who are validators, simplify AddPub…

…lisher proposal (#277)

* add query endpoint for subcribers who are validators

* update somm_proto

* log error rather than panic on invalid subscriber

* allow updates to publishers via governance

proto/pubsub/v1/query.proto

@@ -30,6 +30,10 @@ service Query {
     option (google.api.http).get = "/sommelier/pubsub/v1/subscribers";
   }
 
+  rpc QueryValidatorSubscribers(QueryValidatorSubscribersRequest) returns (QueryValidatorSubscribersResponse) {
+    option (google.api.http).get = "/sommelier/pubsub/v1/validator_subscribers";
+  }
+
   rpc QueryPublisherIntent(QueryPublisherIntentRequest) returns (QueryPublisherIntentResponse) {
     option (google.api.http).get = "/sommelier/pubsub/v1/publisher_intents/{publisher_domain}/{subscription_id}";
   }
@@ -109,6 +113,12 @@ message QuerySubscribersResponse {
   repeated Subscriber subscribers = 1;
 }
 
+message QueryValidatorSubscribersRequest {}
+
+message QueryValidatorSubscribersResponse {
+  repeated Subscriber subscribers = 1;
+}
+
 message QueryPublisherIntentRequest {
   string publisher_domain = 1;
   string subscription_id = 2;

somm_proto/src/prost/pubsub.v1.rs

@@ -459,6 +459,13 @@ pub struct QuerySubscribersResponse {
     pub subscribers: ::prost::alloc::vec::Vec<Subscriber>,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct QueryValidatorSubscribersRequest {}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct QueryValidatorSubscribersResponse {
+    #[prost(message, repeated, tag = "1")]
+    pub subscribers: ::prost::alloc::vec::Vec<Subscriber>,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct QueryPublisherIntentRequest {
     #[prost(string, tag = "1")]
     pub publisher_domain: ::prost::alloc::string::String,
@@ -666,6 +673,22 @@ pub mod query_client {
             let path = http::uri::PathAndQuery::from_static("/pubsub.v1.Query/QuerySubscribers");
             self.inner.unary(request.into_request(), path, codec).await
         }
+        pub async fn query_validator_subscribers(
+            &mut self,
+            request: impl tonic::IntoRequest<super::QueryValidatorSubscribersRequest>,
+        ) -> Result<tonic::Response<super::QueryValidatorSubscribersResponse>, tonic::Status>
+        {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::new(
+                    tonic::Code::Unknown,
+                    format!("Service was not ready: {}", e.into()),
+                )
+            })?;
+            let codec = tonic::codec::ProstCodec::default();
+            let path =
+                http::uri::PathAndQuery::from_static("/pubsub.v1.Query/QueryValidatorSubscribers");
+            self.inner.unary(request.into_request(), path, codec).await
+        }
         pub async fn query_publisher_intent(
             &mut self,
             request: impl tonic::IntoRequest<super::QueryPublisherIntentRequest>,

x/pubsub/client/cli/query.go

@@ -26,6 +26,7 @@ func GetQueryCmd(queryRoute string) *cobra.Command {
 		CmdQueryPublishers(),
 		CmdQuerySubscriber(),
 		CmdQuerySubscribers(),
+		CmdQueryValidatorSubscribers(),
 		CmdQueryPublisherIntent(),
 		CmdQueryPublisherIntents(),
 		CmdQueryPublisherIntentsByPublisherDomain(),
@@ -186,6 +187,34 @@ func CmdQuerySubscribers() *cobra.Command {
 	return cmd
 }
 
+func CmdQueryValidatorSubscribers() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "validator-subscribers",
+		Args:  cobra.NoArgs,
+		Short: "Query subscribers who are validators",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx, err := client.GetClientQueryContext(cmd)
+			if err != nil {
+				return err
+			}
+
+			queryClient := types.NewQueryClient(ctx)
+			req := &types.QueryValidatorSubscribersRequest{}
+
+			res, err := queryClient.QueryValidatorSubscribers(cmd.Context(), req)
+			if err != nil {
+				return err
+			}
+
+			return ctx.PrintProto(res)
+		},
+	}
+
+	flags.AddQueryFlagsToCmd(cmd)
+
+	return cmd
+}
+
 func CmdQueryPublisherIntent() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "publisher-intent",

x/pubsub/keeper/keeper.go

@@ -11,6 +11,7 @@ import (
 	storetypes "github.com/cosmos/cosmos-sdk/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	paramtypes "github.com/cosmos/cosmos-sdk/x/params/types"
+	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 	"github.com/peggyjv/sommelier/v7/x/pubsub/types"
 )
 
@@ -164,6 +165,29 @@ func (k Keeper) GetSubscribers(ctx sdk.Context) (subscribers []*types.Subscriber
 	return
 }
 
+func (k Keeper) GetValidatorSubscribers(ctx sdk.Context) (subscribers []*types.Subscriber) {
+	allSubscribers := k.GetSubscribers(ctx)
+	for _, subscriber := range allSubscribers {
+		subscriberAddress, err := sdk.AccAddressFromBech32(subscriber.Address)
+		if err != nil {
+			ctx.Logger().Error("subscriber address %s not valid bech32 but in state", subscriber.Address)
+			continue
+		}
+		var validatorI stakingtypes.ValidatorI
+		if validator := k.gravityKeeper.GetOrchestratorValidatorAddress(ctx, subscriberAddress); validator == nil {
+			validatorI = k.stakingKeeper.Validator(ctx, sdk.ValAddress(subscriberAddress))
+		} else {
+			validatorI = k.stakingKeeper.Validator(ctx, validator)
+		}
+
+		if validatorI != nil {
+			subscribers = append(subscribers, subscriber)
+		}
+	}
+
+	return
+}
+
 func (k Keeper) DeleteSubscriber(ctx sdk.Context, subscriberAddress sdk.AccAddress) {
 	ctx.KVStore(k.storeKey).Delete(types.GetSubscriberKey(subscriberAddress))
 

x/pubsub/keeper/proposal_handler.go

@@ -9,11 +9,6 @@ import (
 
 // HandleAddPublisherProposal is a handler for executing a passed community publisher addition proposal
 func HandleAddPublisherProposal(ctx sdk.Context, k Keeper, p types.AddPublisherProposal) error {
-	_, found := k.GetPublisher(ctx, p.Domain)
-	if found {
-		return errorsmod.Wrapf(types.ErrAlreadyExists, "publisher already exists with domain: %s", p.Domain)
-	}
-
 	publisher := types.Publisher{
 		Domain:  p.Domain,
 		Address: p.Address,

x/pubsub/keeper/query_server.go

@@ -73,6 +73,12 @@ func (k Keeper) QuerySubscribers(c context.Context, _ *types.QuerySubscribersReq
 	}, nil
 }
 
+func (k Keeper) QueryValidatorSubscribers(c context.Context, _ *types.QueryValidatorSubscribersRequest) (*types.QueryValidatorSubscribersResponse, error) {
+	return &types.QueryValidatorSubscribersResponse{
+		Subscribers: k.GetValidatorSubscribers(sdk.UnwrapSDKContext(c)),
+	}, nil
+}
+
 func (k Keeper) QueryPublisherIntent(c context.Context, req *types.QueryPublisherIntentRequest) (*types.QueryPublisherIntentResponse, error) {
 	if req == nil {
 		return nil, status.Error(codes.InvalidArgument, "empty request")