feat(delegators): add --min-shares / --max-shares filters

Author: ccamel

cmd/delegators.go

@@ -4,26 +4,27 @@ import (
 	"github.com/axone-protocol/cosmos-extractor/pkg/delegators"
 	"github.com/spf13/cobra"
 	"github.com/teambenny/goetl"
+
+	"cosmossdk.io/math"
 )
 
 const (
-	flagHrp = "hrp"
+	flagHrp       = "hrp"
+	flagMinShares = "min-shares"
+	flagMaxShares = "max-shares"
 )
 
 var extractDelegatorsCmd = &cobra.Command{
 	Use:   "delegators [source]",
 	Short: "Extract all delegators",
 	Args:  cobra.ExactArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
-		chainName, _ := cmd.Flags().GetString(flagChainName)
-		src := args[0]
-
-		processors := []goetl.Processor{}
-
-		read, err := delegators.NewDelegatorsReader(chainName, src, logger)
+		read, err := newDelegatorsReader(cmd, args)
 		if err != nil {
 			return err
 		}
+
+		processors := []goetl.Processor{}
 		processors = append(processors, read)
 
 		hrps, err := cmd.Flags().GetStringSlice(flagHrp)
@@ -52,13 +53,56 @@ var extractDelegatorsCmd = &cobra.Command{
 	},
 }
 
+func newDelegatorsReader(cmd *cobra.Command, args []string) (goetl.Processor, error) {
+	chainName, _ := cmd.Flags().GetString(flagChainName)
+	src := args[0]
+
+	delegatorsReaderOpts := []delegators.ReaderOption{
+		delegators.WithChainName(chainName),
+		delegators.WithLogger(logger),
+	}
+
+	v, err := getShares(cmd, flagMinShares)
+	if err != nil {
+		return nil, err
+	}
+	if !v.IsNil() {
+		delegatorsReaderOpts = append(delegatorsReaderOpts, delegators.WithMinSharesFilter(v))
+	}
+
+	v, err = getShares(cmd, flagMaxShares)
+	if err != nil {
+		return nil, err
+	}
+
+	if !v.IsNil() {
+		delegatorsReaderOpts = append(delegatorsReaderOpts, delegators.WithMaxSharesFilter(v))
+	}
+
+	return delegators.NewDelegatorsReader(src, delegatorsReaderOpts...)
+}
+
+func getShares(cmd *cobra.Command, flag string) (math.LegacyDec, error) {
+	shares, err := cmd.Flags().GetString(flag)
+	if err != nil {
+		return math.LegacyDec{}, err
+	}
+	if shares == "" {
+		return math.LegacyDec{}, nil
+	}
+	return math.LegacyNewDecFromStr(shares)
+}
+
 func init() {
 	extractCmd.AddCommand(extractDelegatorsCmd)
 
 	extractDelegatorsCmd.Flags().StringSliceP(
 		flagHrp,
 		"p",
 		[]string{},
-		"One or more Human-Readable Parts (HRPs) to append delegator addresses in the given Bech32 formats (e.g., cosmos, osmo). "+
+		"one or more Human-Readable Parts (HRPs) to append delegator addresses in the given Bech32 formats (e.g., cosmos, osmo). "+
 			"Can be used multiple times for different HRPs.")
+
+	extractDelegatorsCmd.Flags().String(flagMinShares, "", "filter delegators with minimum shares")
+	extractDelegatorsCmd.Flags().String(flagMaxShares, "", "filter delegators with maximum shares")
 }

cmd/extract.go

@@ -21,8 +21,8 @@ var extractCmd = &cobra.Command{
 func init() {
 	rootCmd.AddCommand(extractCmd)
 
-	extractCmd.PersistentFlags().StringP(flagChainName, "n", "cosmos", "Name of the chain")
-	extractCmd.PersistentFlags().StringP(flagOutput, "o", "", "Output file (defaults to stdout)")
+	extractCmd.PersistentFlags().StringP(flagChainName, "n", "cosmos", "name of the chain")
+	extractCmd.PersistentFlags().StringP(flagOutput, "o", "", "output file (defaults to stdout)")
 }
 
 func newCSVWriter(cmd *cobra.Command, _ []string) (goetl.Processor, error) {

cmd/root.go

@@ -42,5 +42,5 @@ func Execute() {
 }
 
 func init() {
-	rootCmd.PersistentFlags().String(flagLogLevel, "info", "The logging level (trace|debug|info|warn|error|fatal|panic)")
+	rootCmd.PersistentFlags().String(flagLogLevel, "info", "logging level (trace|debug|info|warn|error|fatal|panic)")
 }

cmd/version.go

@@ -52,6 +52,6 @@ var versionCmd = &cobra.Command{
 func init() {
 	rootCmd.AddCommand(versionCmd)
 
-	versionCmd.Flags().Bool(flagLong, false, "Print long version information")
-	versionCmd.Flags().StringP(flagFormat, "f", "text", "Output format (text|json)")
+	versionCmd.Flags().Bool(flagLong, false, "print long version information")
+	versionCmd.Flags().StringP(flagFormat, "f", "text", "output format (text|json)")
 }

pkg/delegators/reader.go

@@ -23,20 +23,60 @@ import (
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 )
 
+type ReaderOption func(*delegatorsReader) error
+
+func WithChainName(chainName string) ReaderOption {
+	return func(r *delegatorsReader) error {
+		r.chainName = chainName
+		return nil
+	}
+}
+
+func WithLogger(logger log.Logger) ReaderOption {
+	return func(r *delegatorsReader) error {
+		r.logger = logger
+		return nil
+	}
+}
+
+func WithMinSharesFilter(minShares math.LegacyDec) ReaderOption {
+	return func(r *delegatorsReader) error {
+		r.minSharesFilter = minShares
+		return nil
+	}
+}
+
+func WithMaxSharesFilter(maxShares math.LegacyDec) ReaderOption {
+	return func(r *delegatorsReader) error {
+		r.maxSharesFilter = maxShares
+		return nil
+	}
+}
+
 type delegatorsReader struct {
-	chainName string
-	src       string
-	logger    log.Logger
-	closer    io.Closer
+	chainName       string
+	src             string
+	logger          log.Logger
+	closer          io.Closer
+	minSharesFilter math.LegacyDec
+	maxSharesFilter math.LegacyDec
 }
 
 // NewDelegatorsReader returns a new Reader that reads delegators from a blockchain data stores.
-func NewDelegatorsReader(chainName, src string, logger log.Logger) (goetl.Processor, error) {
-	return &delegatorsReader{
-		chainName: chainName,
+func NewDelegatorsReader(src string, options ...ReaderOption) (goetl.Processor, error) {
+	r := &delegatorsReader{
+		chainName: "mystery",
 		src:       src,
-		logger:    logger,
-	}, nil
+		logger:    log.NewNopLogger(),
+	}
+
+	for _, option := range options {
+		if err := option(r); err != nil {
+			return nil, err
+		}
+	}
+
+	return r, nil
 }
 
 func (r *delegatorsReader) ProcessData(_ etldata.Payload, outputChan chan etldata.Payload, killChan chan error) {
@@ -68,7 +108,13 @@ func (r *delegatorsReader) ProcessData(_ etldata.Payload, outputChan chan etldat
 
 	err = iterateAllAddresses(ctx, keepers.Bank, func(addr sdk.AccAddress) (stop bool) {
 		delegations := lo.RejectMap(validators,
-			toDelegations(ctx, addr, r.logger, keepers.Staking, killChan))
+			extractDelegations(ctx, addr, r.logger, keepers.Staking, killChan))
+		shares := lo.Reduce(delegations, computeShares(), math.LegacyZeroDec())
+
+		if (!r.maxSharesFilter.IsNil() && shares.GT(r.maxSharesFilter)) ||
+			(!r.minSharesFilter.IsNil() && shares.LT(r.minSharesFilter)) {
+			return false
+		}
 
 		for _, delegation := range delegations {
 			payload := Delegation{
@@ -127,7 +173,7 @@ func iterateAllAddresses(ctx context.Context, bankKeeper bankkeeper.BaseKeeper,
 	return err
 }
 
-func toDelegations(
+func extractDelegations(
 	ctx context.Context, address sdk.AccAddress, logger log.Logger, stakingKeeper *stakingkeeper.Keeper, killChan chan error,
 ) func(item stakingtypes.Validator, index int) (stakingtypes.Delegation, bool) {
 	return func(item stakingtypes.Validator, _ int) (stakingtypes.Delegation, bool) {
@@ -152,6 +198,12 @@ func toDelegations(
 	}
 }
 
+func computeShares() func(acc math.LegacyDec, delegation stakingtypes.Delegation, _ int) math.LegacyDec {
+	return func(acc math.LegacyDec, delegation stakingtypes.Delegation, _ int) math.LegacyDec {
+		return acc.Add(delegation.Shares)
+	}
+}
+
 func guessPrefixFromValoper(valoper string) (string, error) {
 	if idx := strings.Index(valoper, "valoper"); idx != -1 {
 		return valoper[:idx], nil