Deps/update (#2)

* Updated go and some other dependencies

* Updated indexer AttachTo; Added MustInput and MustOutput methods

* Removed toolchain from go.mod

* Updated go ver to 1.22 in GH test workflow

* GH linter update

* Added Indexer inheritance from modules.BaseModule

* Fixed Docker critical vulnerability

* Added module.BaseModule to Indexer; Added db healthcheck DB and user

* Fix: indexer module

---------

Co-authored-by: Artem <[email protected]>

.github/workflows/test.yml

@@ -11,20 +11,20 @@ jobs:
     steps:
       - uses: actions/setup-go@v3
         with:
-          go-version: '1.20'
+          go-version: '1.22'
       - uses: actions/checkout@v3
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v6
         with:
-          version: v1.51.2
+          version: v1.60
           args: --timeout=4m
   test:
     runs-on: ubuntu-latest
     steps:
       - name: install Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v5
         with:
-          go-version: 1.20.x
+          go-version: 1.22.5
       - name: checkout code
         uses: actions/checkout@v2
       - uses: actions/cache@v2

build/dipdup.yml

@@ -115,7 +115,7 @@ database:
   port: ${POSTGRES_PORT:-5432}
   user: ${POSTGRES_USER:-dipdup}
   password: ${POSTGRES_PASSWORD:-changeme}
-  database: ${POSTGRES_DB:-dipdup}
+  database: ${POSTGRES_DB:-starknet_metadata}
 
 hasura:
   url: http://${HASURA_HOST:-hasura}:${HASURA_PORT:-8080}

build/metadata/Dockerfile

@@ -1,7 +1,7 @@
 # ---------------------------------------------------------------------
 #  The first stage container, for building the application
 # ---------------------------------------------------------------------
-FROM golang:1.20-alpine as builder
+FROM golang:1.22.5-alpine as builder
 
 ENV CGO_ENABLED=0
 ENV GO111MODULE=on

cmd/metadata/indexer.go

@@ -24,14 +24,16 @@ var (
 
 // input name
 const (
-	InputName = "input"
+	InputName  = "input"
+	OutputName = "output"
 )
 
 // Indexer -
 type Indexer struct {
+	modules.BaseModule
+
 	client       *grpc.Client
 	storage      postgres.Storage
-	input        *modules.Input
 	state        *models.State
 	subscription grpc.Subscription
 	subId        uint64
@@ -46,12 +48,11 @@ type Indexer struct {
 // NewIndexer -
 func NewIndexer(cfg Metadata, datasources map[string]config.DataSource, pg postgres.Storage, client *grpc.Client, ipfsNode *ipfs.Node) (*Indexer, error) {
 	indexer := &Indexer{
-		client:  client,
-		storage: pg,
-
-		state: new(models.State),
-		input: modules.NewInput(InputName),
-		wg:    new(sync.WaitGroup),
+		client:     client,
+		storage:    pg,
+		BaseModule: modules.New("Indexer"),
+		state:      new(models.State),
+		wg:         new(sync.WaitGroup),
 	}
 	indexer.channel = NewChannel(pg, indexer.state)
 	filler, err := NewFiller(cfg.Filler, datasources, pg.TokenMetadata, client)
@@ -61,6 +62,9 @@ func NewIndexer(cfg Metadata, datasources map[string]config.DataSource, pg postg
 	indexer.filler = filler
 	indexer.receiver = NewReceiver(cfg.Receiver, pg.TokenMetadata, ipfsNode)
 
+	indexer.CreateInputWithCapacity(InputName, 1024*16)
+	indexer.CreateOutput(OutputName)
+
 	return indexer, nil
 }
 
@@ -83,11 +87,6 @@ func (indexer *Indexer) Start(ctx context.Context) {
 	indexer.receiver.Start(ctx)
 }
 
-// Name -
-func (indexer *Indexer) Name() string {
-	return "starknet_metadata_indexer"
-}
-
 // Subscribe -
 func (indexer *Indexer) Subscribe(ctx context.Context, subscriptions map[string]grpc.Subscription) error {
 	s, ok := subscriptions["metadata"]
@@ -126,16 +125,6 @@ func (indexer *Indexer) init(ctx context.Context) error {
 	}
 }
 
-// Input - returns input by name
-func (indexer *Indexer) Input(name string) (*modules.Input, error) {
-	switch name {
-	case InputName:
-		return indexer.input, nil
-	default:
-		return nil, errors.Wrap(modules.ErrUnknownInput, name)
-	}
-}
-
 func (indexer *Indexer) listen(ctx context.Context) {
 	defer indexer.wg.Done()
 
@@ -153,7 +142,7 @@ func (indexer *Indexer) listen(ctx context.Context) {
 
 		case msg, ok := <-input.Listen():
 			if !ok {
-				continue
+				return
 			}
 
 			switch typ := msg.(type) {
@@ -186,7 +175,7 @@ func (indexer *Indexer) reconnectThread(ctx context.Context) {
 	}
 }
 
-func (indexer *Indexer) resubscribe(ctx context.Context, id uint64) error {
+func (indexer *Indexer) resubscribe(ctx context.Context, _ uint64) error {
 	for !indexer.channel.IsEmpty() {
 		select {
 		case <-ctx.Done():
@@ -233,21 +222,6 @@ func (indexer *Indexer) actualFilters(ctx context.Context, sub *grpc.Subscriptio
 	return nil
 }
 
-// Output - returns output by name
-func (indexer *Indexer) Output(name string) (*modules.Output, error) {
-	return nil, errors.Wrap(modules.ErrUnknownOutput, name)
-}
-
-// AttachTo - attach input to output with name
-func (indexer *Indexer) AttachTo(name string, input *modules.Input) error {
-	output, err := indexer.Output(name)
-	if err != nil {
-		return err
-	}
-	output.Attach(input)
-	return nil
-}
-
 // Unsubscribe -
 func (indexer *Indexer) Unsubscribe(ctx context.Context) error {
 	log.Info().Uint64("id", indexer.subId).Msg("unsubscribing...")
@@ -273,9 +247,5 @@ func (indexer *Indexer) Close() error {
 		return err
 	}
 
-	if err := indexer.input.Close(); err != nil {
-		return err
-	}
-
 	return nil
 }

cmd/metadata/main.go

@@ -13,9 +13,7 @@ import (
 	"github.com/dipdup-io/starknet-metadata/internal/storage"
 	"github.com/dipdup-io/starknet-metadata/internal/storage/postgres"
 	"github.com/dipdup-net/go-lib/hasura"
-	"github.com/dipdup-net/indexer-sdk/pkg/modules"
 	grpcSDK "github.com/dipdup-net/indexer-sdk/pkg/modules/grpc"
-	"github.com/dipdup-net/indexer-sdk/pkg/modules/printer"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
@@ -82,7 +80,7 @@ func main() {
 		return
 	}
 
-	views, err := createViews(ctx, pg)
+	views, err := createViews(pg)
 	if err != nil {
 		log.Panic().Err(err).Msg("create views")
 		return
@@ -117,8 +115,8 @@ func main() {
 		return
 	}
 
-	if err := modules.Connect(client, indexer, grpc.OutputMessages, printer.InputName); err != nil {
-		log.Panic().Err(err).Msg("module connect")
+	if err := indexer.AttachTo(client, grpc.OutputMessages, InputName); err != nil {
+		log.Panic().Err(err).Msg("connect indexer to grpc module")
 		return
 	}
 

cmd/metadata/views.go

@@ -1,15 +1,14 @@
 package main
 
 import (
-	"context"
 	"fmt"
 	"os"
 	"strings"
 
 	"github.com/dipdup-io/starknet-metadata/internal/storage/postgres"
 )
 
-func createViews(ctx context.Context, strg postgres.Storage) ([]string, error) {
+func createViews(strg postgres.Storage) ([]string, error) {
 	files, err := os.ReadDir("views")
 	if err != nil {
 		return nil, err

docker-compose.yml

@@ -29,7 +29,7 @@ services:
       - POSTGRES_DB=${POSTGRES_DB:-starknet_metadata}
       - POSTGRES_PASSWORD=${POSTGRES_PASSWORD:-changeme}
     healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U dipdup"]
+      test: [ "CMD-SHELL", "pg_isready -d $${POSTGRES_DB} -U $${POSTGRES_USER}" ]
       interval: 10s
       timeout: 5s
       retries: 5