adding select with limit

config.yaml

@@ -16,3 +16,6 @@ key:
 
 cosmos:
   addr: "localhost:9090"
+
+select:
+  limit: 20

internal/config/main.go

@@ -13,6 +13,7 @@ type Config interface {
 	pgdb.Databaser
 	Keyer
 	Cosmoser
+	Selecter
 
 	Storage() data.Storage
 }
@@ -23,6 +24,7 @@ type config struct {
 	pgdb.Databaser
 	Keyer
 	Cosmoser
+	Selecter
 
 	getter  kv.Getter
 	storage comfig.Once
@@ -38,5 +40,6 @@ func New(getter kv.Getter) Config {
 		Databaser:  pgdb.NewDatabaser(getter),
 		Keyer:      NewKeyer(getter),
 		Cosmoser:   NewCosmoser(getter),
+		Selecter:   NewSelecter(getter),
 	}
 }

internal/config/select.go

@@ -0,0 +1,36 @@
+package config
+
+import (
+	"gitlab.com/distributed_lab/figure"
+	"gitlab.com/distributed_lab/kit/comfig"
+	"gitlab.com/distributed_lab/kit/kv"
+)
+
+type Selecter interface {
+	Limit() uint64
+}
+
+type selecter struct {
+	getter kv.Getter
+	once   comfig.Once
+}
+
+func NewSelecter(getter kv.Getter) Selecter {
+	return &selecter{
+		getter: getter,
+	}
+}
+
+func (s *selecter) Limit() uint64 {
+	return s.once.Do(func() interface{} {
+		var config struct {
+			Limit uint64 `fig:"limit"`
+		}
+
+		if err := figure.Out(&config).From(kv.MustGetStringMap(s.getter, "select")).Please(); err != nil {
+			panic(err)
+		}
+
+		return config.Limit
+	}).(uint64)
+}

internal/data/main.go

@@ -20,5 +20,6 @@ type TransactionQ interface {
 	InsertCtx(ctx context.Context, t *Transaction) error
 	DeleteCtx(ctx context.Context, t *Transaction) error
 	Select(ctx context.Context) ([]Transaction, error)
+	SelectWithLimit(ctx context.Context, limit uint64) ([]Transaction, error)
 }
 type GorpMigrationQ interface{}

internal/data/pg/transaction.go

@@ -13,3 +13,10 @@ func (q TransactionQ) Select(ctx context.Context) ([]data.Transaction, error) {
 	err := q.db.SelectContext(ctx, &result, stmt)
 	return result, errors.Wrap(err, "failed to exec stmt")
 }
+
+func (q TransactionQ) SelectWithLimit(ctx context.Context, limit uint64) ([]data.Transaction, error) {
+	stmt := squirrel.Select(colsTransaction).From("transactions").Limit(limit)
+	var result []data.Transaction
+	err := q.db.SelectContext(ctx, &result, stmt)
+	return result, errors.Wrap(err, "failed to exec stmt")
+}

internal/services/broadcaster/broadcaster.go

@@ -38,6 +38,7 @@ type broadcaster struct {
 	txclient txclient.ServiceClient
 	auth     authtypes.QueryClient
 	txQ      data.TransactionQ
+	limit    uint64
 }
 
 func Run(ctx context.Context, cfg config.Config) {
@@ -53,13 +54,14 @@ func Run(ctx context.Context, cfg config.Config) {
 			codec.NewProtoCodec(codectypes.NewInterfaceRegistry()),
 			[]signing.SignMode{signing.SignMode_SIGN_MODE_DIRECT},
 		),
+		limit: cfg.Limit(),
 	}
 
 	running.WithBackOff(ctx, svc.log, "run", svc.runOnceIndexing, 5*time.Second, 5*time.Second, 5*time.Second)
 }
 
 func (t *broadcaster) runOnceIndexing(ctx context.Context) error {
-	txs, err := t.txQ.Select(ctx)
+	txs, err := t.txQ.SelectWithLimit(ctx, t.limit)
 	if err != nil {
 		return errors.Wrap(err, "failed to select txs")
 	}