From 1478a02342470b4ddc59e6d3b324690aee31a2d7 Mon Sep 17 00:00:00 2001
From: mismirnov <mi.smirnov24@gmail.com>
Date: Mon, 16 Dec 2024 23:02:23 +0300
Subject: [PATCH] Datasources api WIP;

---
 .run/API.run.xml                 |  33 ++++++-
 .run/Indexer.run.xml             |  29 ++++++
 cmd/tvl/init.go                  |  69 ++++++++++++++
 cmd/tvl/main.go                  |  52 +++++++++++
 configs/dipdup.yml               |  10 ++
 internal/tvl/l2beat/api.go       | 103 +++++++++++++++++++++
 internal/tvl/l2beat/interface.go |  11 +++
 internal/tvl/l2beat/tvl.go       |  40 ++++++++
 internal/tvl/lama/api.go         | 103 +++++++++++++++++++++
 internal/tvl/lama/interface.go   |  11 +++
 internal/tvl/lama/tvl.go         |  24 +++++
 pkg/tvl/indexer.go               | 153 +++++++++++++++++++++++++++++++
 pkg/tvl/receiver/receiver.go     |  46 ++++++++++
 13 files changed, 682 insertions(+), 2 deletions(-)
 create mode 100644 cmd/tvl/init.go
 create mode 100644 cmd/tvl/main.go
 create mode 100644 internal/tvl/l2beat/api.go
 create mode 100644 internal/tvl/l2beat/interface.go
 create mode 100644 internal/tvl/l2beat/tvl.go
 create mode 100644 internal/tvl/lama/api.go
 create mode 100644 internal/tvl/lama/interface.go
 create mode 100644 internal/tvl/lama/tvl.go
 create mode 100644 pkg/tvl/indexer.go
 create mode 100644 pkg/tvl/receiver/receiver.go

diff --git a/.run/API.run.xml b/.run/API.run.xml
index 2704c9a5..ed152cf0 100644
--- a/.run/API.run.xml
+++ b/.run/API.run.xml
@@ -1,8 +1,37 @@
 <component name="ProjectRunConfigurationManager">
   <configuration default="false" name="API" type="GoApplicationRunConfiguration" factoryName="Go Application">
     <module name="celestia-indexer" />
-    <working_directory value="$PROJECT_DIR$" />
-    <parameters value="-c configs/dipdup.yml" />
+    <working_directory value="$PROJECT_DIR$/cmd/api" />
+    <parameters value="-c ../../configs/dipdup.yml" />
+    <envs>
+      <env name="API_PROMETHEUS_ENABLED" value="false" />
+      <env name="API_RATE_LIMIT" value="20" />
+      <env name="API_REQUEST_TIMEOUT" value="10" />
+      <env name="API_WEBSOCKET_ENABLED" value="true" />
+      <env name="BINANCE_API_RPS" value="5" />
+      <env name="BINANCE_API_TIMEOUT" value="10" />
+      <env name="BINANCE_API_URL" value="https://api.binance.com/" />
+      <env name="CELENIUM_ENV" value="production" />
+      <env name="CELESTIA_DAL_API_RPS" value="10" />
+      <env name="CELESTIA_DAL_API_TIMEOUT" value="30 # seconds" />
+      <env name="CELESTIA_DAL_API_URL" value="" />
+      <env name="CELESTIA_NODE_AUTH_TOKEN" value="" />
+      <env name="CELESTIA_NODE_RPS" value="5" />
+      <env name="CELESTIA_NODE_TIMEOUT" value="10 # seconds" />
+      <env name="CELESTIA_NODE_URL" value="https://node.celenium.io" />
+      <env name="CELESTIA_NODE_WS_URL" value="https://node.celenium.io:443" />
+      <env name="INDEXER_BLOCK_PERIOD" value="15 # seconds" />
+      <env name="INDEXER_NAME" value="celestia_indexer" />
+      <env name="INDEXER_SCRIPTS_DIR" value="./../../database" />
+      <env name="INDEXER_START_LEVEL" value="1" />
+      <env name="INDEXER_THREADS_COUNT" value="10" />
+      <env name="LOG_LEVEL" value="debug" />
+      <env name="POSTGRES_DB" value="celestia" />
+      <env name="POSTGRES_HOST" value="localhost" />
+      <env name="POSTGRES_PASSWORD" value="changeme" />
+      <env name="POSTGRES_PORT" value="5432" />
+      <env name="POSTGRES_USER" value="dipdup" />
+    </envs>
     <EXTENSION ID="net.ashald.envfile">
       <option name="IS_ENABLED" value="true" />
       <option name="IS_SUBST" value="false" />
diff --git a/.run/Indexer.run.xml b/.run/Indexer.run.xml
index 844f81bd..bbbede97 100644
--- a/.run/Indexer.run.xml
+++ b/.run/Indexer.run.xml
@@ -3,6 +3,35 @@
     <module name="celestia-indexer" />
     <working_directory value="$PROJECT_DIR$" />
     <parameters value="-c configs/dipdup.yml" />
+    <envs>
+      <env name="API_PROMETHEUS_ENABLED" value="false" />
+      <env name="API_RATE_LIMIT" value="20" />
+      <env name="API_REQUEST_TIMEOUT" value="10" />
+      <env name="API_WEBSOCKET_ENABLED" value="true" />
+      <env name="BINANCE_API_RPS" value="5" />
+      <env name="BINANCE_API_TIMEOUT" value="10" />
+      <env name="BINANCE_API_URL" value="https://api.binance.com/" />
+      <env name="CELENIUM_ENV" value="production" />
+      <env name="CELESTIA_DAL_API_RPS" value="10" />
+      <env name="CELESTIA_DAL_API_TIMEOUT" value="30 # seconds" />
+      <env name="CELESTIA_DAL_API_URL" value="" />
+      <env name="CELESTIA_NODE_AUTH_TOKEN" value="" />
+      <env name="CELESTIA_NODE_RPS" value="5" />
+      <env name="CELESTIA_NODE_TIMEOUT" value="10 # seconds" />
+      <env name="CELESTIA_NODE_URL" value="https://node.celenium.io" />
+      <env name="CELESTIA_NODE_WS_URL" value="https://node.celenium.io:443" />
+      <env name="INDEXER_BLOCK_PERIOD" value="15 # seconds" />
+      <env name="INDEXER_NAME" value="celestia_indexer" />
+      <env name="INDEXER_SCRIPTS_DIR" value="./database" />
+      <env name="INDEXER_START_LEVEL" value="1" />
+      <env name="INDEXER_THREADS_COUNT" value="10" />
+      <env name="LOG_LEVEL" value="debug" />
+      <env name="POSTGRES_DB" value="celestia" />
+      <env name="POSTGRES_HOST" value="localhost" />
+      <env name="POSTGRES_PASSWORD" value="changeme" />
+      <env name="POSTGRES_PORT" value="5432" />
+      <env name="POSTGRES_USER" value="dipdup" />
+    </envs>
     <EXTENSION ID="net.ashald.envfile">
       <option name="IS_ENABLED" value="true" />
       <option name="IS_SUBST" value="false" />
diff --git a/cmd/tvl/init.go b/cmd/tvl/init.go
new file mode 100644
index 00000000..180354b4
--- /dev/null
+++ b/cmd/tvl/init.go
@@ -0,0 +1,69 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package main
+
+import (
+	"os"
+	"strconv"
+
+	"github.com/celenium-io/celestia-indexer/pkg/indexer/config"
+	goLibConfig "github.com/dipdup-net/go-lib/config"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+func init() {
+	log.Logger = log.Output(zerolog.ConsoleWriter{
+		Out:        os.Stdout,
+		TimeFormat: "2006-01-02 15:04:05",
+	})
+}
+
+func initConfig() (*config.Config, error) {
+	configPath := rootCmd.PersistentFlags().StringP("config", "c", "dipdup.yml", "path to YAML config file")
+	if err := rootCmd.Execute(); err != nil {
+		log.Panic().Err(err).Msg("command line execute")
+		return nil, err
+	}
+
+	if err := rootCmd.MarkFlagRequired("config"); err != nil {
+		log.Panic().Err(err).Msg("config command line arg is required")
+		return nil, err
+	}
+
+	var cfg config.Config
+	if err := goLibConfig.Parse(*configPath, &cfg); err != nil {
+		log.Panic().Err(err).Msg("parsing config file")
+		return nil, err
+	}
+
+	if cfg.LogLevel == "" {
+		cfg.LogLevel = zerolog.LevelInfoValue
+	}
+
+	return &cfg, nil
+}
+
+func initLogger(level string) error {
+	logLevel, err := zerolog.ParseLevel(level)
+	if err != nil {
+		log.Panic().Err(err).Msg("parsing log level")
+		return err
+	}
+	zerolog.SetGlobalLevel(logLevel)
+	zerolog.CallerMarshalFunc = func(pc uintptr, file string, line int) string {
+		short := file
+		for i := len(file) - 1; i > 0; i-- {
+			if file[i] == '/' {
+				short = file[i+1:]
+				break
+			}
+		}
+		file = short
+		return file + ":" + strconv.Itoa(line)
+	}
+	log.Logger = log.Logger.With().Caller().Logger()
+
+	return nil
+}
diff --git a/cmd/tvl/main.go b/cmd/tvl/main.go
new file mode 100644
index 00000000..0496736b
--- /dev/null
+++ b/cmd/tvl/main.go
@@ -0,0 +1,52 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package main
+
+import (
+	"context"
+	indexer "github.com/celenium-io/celestia-indexer/pkg/tvl"
+	"github.com/dipdup-net/indexer-sdk/pkg/modules/stopper"
+
+	"github.com/rs/zerolog/log"
+	"github.com/spf13/cobra"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+var rootCmd = &cobra.Command{
+	Use:   "indexer",
+	Short: "DipDup Verticals | Celenium TVL Scanner",
+}
+
+func main() {
+	cfg, err := initConfig()
+	if err != nil {
+		return
+	}
+
+	if err = initLogger(cfg.LogLevel); err != nil {
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	notifyCtx, notifyCancel := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM, syscall.SIGINT)
+	defer notifyCancel()
+
+	stopperModule := stopper.NewModule(cancel)
+	indexerModule, err := indexer.New(ctx, *cfg, &stopperModule)
+	if err != nil {
+		log.Panic().Err(err).Msg("error during indexer module creation")
+		return
+	}
+
+	stopperModule.Start(ctx)
+	indexerModule.Start(ctx)
+
+	<-notifyCtx.Done()
+	cancel()
+
+	log.Info().Msg("stopped")
+}
diff --git a/configs/dipdup.yml b/configs/dipdup.yml
index 2c471fbc..adc37f90 100644
--- a/configs/dipdup.yml
+++ b/configs/dipdup.yml
@@ -35,6 +35,16 @@ datasources:
     url: ${BINANCE_API_URL:-https://api.binance.com/}
     rps: ${BINANCE_API_RPS:-5}
     timeout: ${BINANCE_API_TIMEOUT:-10}
+  l2beat:
+    kind: l2beat_api
+    url: ${L2BEAT_API_URL:-https://l2beat.com/api/}
+    rps: ${L2BEAT_API_RPS:-5}
+    timeout: ${L2BEAT_API_TIMEOUT:-10}
+  lama:
+    kind: lama_api
+    url: ${LAMA_API_URL:-https://api.llama.fi/}
+    rps: ${LAMA_API_RPS:-5}
+    timeout: ${LAMA_API_TIMEOUT:-10}
 
 api:
   bind: ${API_HOST:-0.0.0.0}:${API_PORT:-9876}
diff --git a/internal/tvl/l2beat/api.go b/internal/tvl/l2beat/api.go
new file mode 100644
index 00000000..fead9280
--- /dev/null
+++ b/internal/tvl/l2beat/api.go
@@ -0,0 +1,103 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package l2beat
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/dipdup-net/go-lib/config"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+	"golang.org/x/time/rate"
+	"io"
+	"net/http"
+	"net/url"
+	"time"
+)
+
+type API struct {
+	client    *http.Client
+	cfg       config.DataSource
+	rateLimit *rate.Limiter
+	log       zerolog.Logger
+}
+
+func NewAPI(cfg config.DataSource) API {
+	rps := cfg.RequestsPerSecond
+	if cfg.RequestsPerSecond < 1 || cfg.RequestsPerSecond > 100 {
+		rps = 10
+	}
+
+	t := http.DefaultTransport.(*http.Transport).Clone()
+	t.MaxIdleConns = rps
+	t.MaxConnsPerHost = rps
+	t.MaxIdleConnsPerHost = rps
+
+	return API{
+		client: &http.Client{
+			Transport: t,
+		},
+		cfg:       cfg,
+		rateLimit: rate.NewLimiter(rate.Every(time.Second/time.Duration(rps)), rps),
+		log:       log.With().Str("module", "L2Beat api").Logger(),
+	}
+}
+
+func (api *API) get(ctx context.Context, path string, args map[string]string, output any) error {
+	u, err := url.Parse(api.cfg.URL)
+	if err != nil {
+		return err
+	}
+	u.Path, err = url.JoinPath(u.Path, path)
+	if err != nil {
+		return err
+	}
+
+	values := u.Query()
+	for key, value := range args {
+		values.Add(key, value)
+	}
+	u.RawQuery = values.Encode()
+
+	if api.rateLimit != nil {
+		if err := api.rateLimit.Wait(ctx); err != nil {
+			return err
+		}
+	}
+
+	start := time.Now()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
+	if err != nil {
+		return err
+	}
+
+	response, err := api.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer closeWithLogError(response.Body, api.log)
+
+	api.log.Trace().
+		Int64("ms", time.Since(start).Milliseconds()).
+		Str("url", u.String()).
+		Msg("request")
+
+	if response.StatusCode != http.StatusOK {
+		return errors.Errorf("invalid status: %d", response.StatusCode)
+	}
+
+	err = json.NewDecoder(response.Body).Decode(output)
+	return err
+}
+
+func closeWithLogError(stream io.ReadCloser, log zerolog.Logger) {
+	if _, err := io.Copy(io.Discard, stream); err != nil {
+		log.Err(err).Msg("L2Beat api copy GET body response to discard")
+	}
+	if err := stream.Close(); err != nil {
+		log.Err(err).Msg("L2Beat api close GET body request")
+	}
+}
diff --git a/internal/tvl/l2beat/interface.go b/internal/tvl/l2beat/interface.go
new file mode 100644
index 00000000..c608e7a1
--- /dev/null
+++ b/internal/tvl/l2beat/interface.go
@@ -0,0 +1,11 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package l2beat
+
+import "context"
+
+//go:generate mockgen -source=$GOFILE -destination=mock/$GOFILE -package=mock -typed
+type IApi interface {
+	TVL(ctx context.Context, arguments *TVLArgs) (result TVLResponse, err error)
+}
diff --git a/internal/tvl/l2beat/tvl.go b/internal/tvl/l2beat/tvl.go
new file mode 100644
index 00000000..f52e047a
--- /dev/null
+++ b/internal/tvl/l2beat/tvl.go
@@ -0,0 +1,40 @@
+package l2beat
+
+import (
+	"context"
+)
+
+type Data struct {
+	JSON [][]interface{} `json:"json"`
+}
+
+type Result struct {
+	Data Data `json:"data"`
+}
+
+type TVLResponse []struct {
+	Result Result `json:"result"`
+}
+
+type TVLArgs struct {
+	Batch int64    `json:"batch"`
+	Input JsonData `json:"input"`
+}
+
+type JsonData struct {
+	Filter                  Filter `json:"filter"`
+	Range                   string `json:"range"`
+	ExcludeAssociatedTokens bool   `json:"excludeAssociatedTokens"`
+}
+
+type Filter struct {
+	Type       string   `json:"type"`
+	ProjectIds []string `json:"projectIds"`
+}
+
+func (api API) TVL(ctx context.Context, arguments *TVLArgs) (result TVLResponse, err error) {
+	// TODO: cast args to query params
+	args := map[string]string{}
+	err = api.get(ctx, "trpc/tvl.chart", args, &result)
+	return
+}
diff --git a/internal/tvl/lama/api.go b/internal/tvl/lama/api.go
new file mode 100644
index 00000000..72a19afb
--- /dev/null
+++ b/internal/tvl/lama/api.go
@@ -0,0 +1,103 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package lama
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/dipdup-net/go-lib/config"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+	"golang.org/x/time/rate"
+	"io"
+	"net/http"
+	"net/url"
+	"time"
+)
+
+type API struct {
+	client    *http.Client
+	cfg       config.DataSource
+	rateLimit *rate.Limiter
+	log       zerolog.Logger
+}
+
+func NewAPI(cfg config.DataSource) API {
+	rps := cfg.RequestsPerSecond
+	if cfg.RequestsPerSecond < 1 || cfg.RequestsPerSecond > 100 {
+		rps = 10
+	}
+
+	t := http.DefaultTransport.(*http.Transport).Clone()
+	t.MaxIdleConns = rps
+	t.MaxConnsPerHost = rps
+	t.MaxIdleConnsPerHost = rps
+
+	return API{
+		client: &http.Client{
+			Transport: t,
+		},
+		cfg:       cfg,
+		rateLimit: rate.NewLimiter(rate.Every(time.Second/time.Duration(rps)), rps),
+		log:       log.With().Str("module", "DeFi Lama api").Logger(),
+	}
+}
+
+func (api *API) get(ctx context.Context, path string, args map[string]string, output any) error {
+	u, err := url.Parse(api.cfg.URL)
+	if err != nil {
+		return err
+	}
+	u.Path, err = url.JoinPath(u.Path, path)
+	if err != nil {
+		return err
+	}
+
+	values := u.Query()
+	for key, value := range args {
+		values.Add(key, value)
+	}
+	u.RawQuery = values.Encode()
+
+	if api.rateLimit != nil {
+		if err := api.rateLimit.Wait(ctx); err != nil {
+			return err
+		}
+	}
+
+	start := time.Now()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
+	if err != nil {
+		return err
+	}
+
+	response, err := api.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer closeWithLogError(response.Body, api.log)
+
+	api.log.Trace().
+		Int64("ms", time.Since(start).Milliseconds()).
+		Str("url", u.String()).
+		Msg("request")
+
+	if response.StatusCode != http.StatusOK {
+		return errors.Errorf("invalid status: %d", response.StatusCode)
+	}
+
+	err = json.NewDecoder(response.Body).Decode(output)
+	return err
+}
+
+func closeWithLogError(stream io.ReadCloser, log zerolog.Logger) {
+	if _, err := io.Copy(io.Discard, stream); err != nil {
+		log.Err(err).Msg("DeFi Lama api copy GET body response to discard")
+	}
+	if err := stream.Close(); err != nil {
+		log.Err(err).Msg("DeFi Lama api close GET body request")
+	}
+}
diff --git a/internal/tvl/lama/interface.go b/internal/tvl/lama/interface.go
new file mode 100644
index 00000000..3af03b45
--- /dev/null
+++ b/internal/tvl/lama/interface.go
@@ -0,0 +1,11 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package lama
+
+import "context"
+
+//go:generate mockgen -source=$GOFILE -destination=mock/$GOFILE -package=mock -typed
+type IApi interface {
+	TVL(ctx context.Context, arguments *TVLArgs) (result []TVLResponse, err error)
+}
diff --git a/internal/tvl/lama/tvl.go b/internal/tvl/lama/tvl.go
new file mode 100644
index 00000000..8a1c2428
--- /dev/null
+++ b/internal/tvl/lama/tvl.go
@@ -0,0 +1,24 @@
+package lama
+
+import (
+	"context"
+	"net/url"
+)
+
+type TVLResponse struct {
+	Date int64   `json:"date"`
+	TVL  float64 `json:"tvl"`
+}
+
+type TVLArgs struct {
+	Chain string `json:"chain"`
+}
+
+func (api API) TVL(ctx context.Context, arguments *TVLArgs) (result []TVLResponse, err error) {
+	path, err := url.JoinPath("v2/historicalChainTvl", arguments.Chain)
+	if err != nil {
+		return nil, err
+	}
+	err = api.get(ctx, path, nil, &result)
+	return
+}
diff --git a/pkg/tvl/indexer.go b/pkg/tvl/indexer.go
new file mode 100644
index 00000000..9c535cec
--- /dev/null
+++ b/pkg/tvl/indexer.go
@@ -0,0 +1,153 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package tvl
+
+import (
+	"context"
+	"github.com/celenium-io/celestia-indexer/internal/tvl/l2beat"
+	"github.com/celenium-io/celestia-indexer/internal/tvl/lama"
+	"github.com/celenium-io/celestia-indexer/pkg/indexer/config"
+	"github.com/celenium-io/celestia-indexer/pkg/indexer/parser"
+	"github.com/celenium-io/celestia-indexer/pkg/indexer/storage"
+	"github.com/celenium-io/celestia-indexer/pkg/tvl/receiver"
+	"github.com/dipdup-net/indexer-sdk/pkg/modules"
+	"github.com/dipdup-net/indexer-sdk/pkg/modules/stopper"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+type Indexer struct {
+	cfg       config.Config
+	l2beatApi l2beat.API
+	lamaApi   lama.API
+	receiver  *receiver.Module
+	//parser    *parser.Module
+	//storage   *storage.Module
+	stopper modules.Module
+	log     zerolog.Logger
+}
+
+func New(ctx context.Context, cfg config.Config, stopperModule modules.Module) (Indexer, error) {
+	l2beatApi, lamaApi, r, err := createReceiver(cfg)
+	if err != nil {
+		return Indexer{}, errors.Wrap(err, "while creating receiver module")
+	}
+
+	//p, err := createParser(r)
+	//if err != nil {
+	//	return Indexer{}, errors.Wrap(err, "while creating parser module")
+	//}
+	//
+	//s, err := createStorage(pg, cfg, p)
+	//if err != nil {
+	//	return Indexer{}, errors.Wrap(err, "while creating storage module")
+	//}
+
+	//err = attachStopper(stopperModule, r, p, s)
+	//if err != nil {
+	//	return Indexer{}, errors.Wrap(err, "while creating stopper module")
+	//}
+
+	return Indexer{
+		cfg:       cfg,
+		l2beatApi: l2beatApi,
+		lamaApi:   lamaApi,
+		receiver:  r,
+		//parser:   p,
+		//storage:  s,
+		//stopper:  stopperModule,
+		log: log.With().Str("module", "TVL indexer").Logger(),
+	}, nil
+}
+
+func (i *Indexer) Start(ctx context.Context) {
+	i.log.Info().Msg("starting...")
+
+	//i.storage.Start(ctx)
+	//i.parser.Start(ctx)
+	i.receiver.Start(ctx)
+}
+
+func (i *Indexer) Close() error {
+	i.log.Info().Msg("closing...")
+
+	if err := i.receiver.Close(); err != nil {
+		log.Err(err).Msg("closing receiver")
+	}
+
+	//if err := i.parser.Close(); err != nil {
+	//	log.Err(err).Msg("closing parser")
+	//}
+	//if err := i.storage.Close(); err != nil {
+	//	log.Err(err).Msg("closing storage")
+	//}
+
+	if err := i.stopper.Close(); err != nil {
+		log.Err(err).Msg("closing stopper")
+	}
+
+	return nil
+}
+
+func createReceiver(cfg config.Config) (l2beat.API, lama.API, *receiver.Module, error) {
+	l2beatDs, ok := cfg.DataSources["l2beat"]
+
+	if !ok {
+		return l2beat.API{}, lama.API{}, nil, errors.New("can't find L2Beat api datasource")
+	}
+
+	lamaDs, ok := cfg.DataSources["lama"]
+
+	if !ok {
+		return l2beat.API{}, lama.API{}, nil, errors.New("can't find DeFi Lama api datasource")
+	}
+
+	l2beatApi := l2beat.NewAPI(l2beatDs)
+	lamaApi := lama.NewAPI(lamaDs)
+	receiverModule := receiver.NewModule(l2beatApi, lamaApi)
+
+	return l2beatApi, lamaApi, &receiverModule, nil
+}
+
+//func createParser(receiverModule modules.Module) (*parser.Module, error) {
+//	parserModule := parser.NewModule()
+//
+//	if err := parserModule.AttachTo(receiverModule, receiver.DataOutput, parser.InputName); err != nil {
+//		return nil, errors.Wrap(err, "while attaching parser to receiver")
+//	}
+//
+//	return &parserModule, nil
+//}
+//
+//func createStorage(cfg config.Config, parserModule modules.Module) (*storage.Module, error) {
+//	storageModule := storage.NewModule(pg.Transactable, cfg.Indexer)
+//
+//	if err := storageModule.AttachTo(parserModule, parser.OutputName, storage.DataInput); err != nil {
+//		return nil, errors.Wrap(err, "while attaching storage to parser")
+//	}
+//
+//	return &storageModule, nil
+//}
+
+func attachStopper(
+	stopperModule modules.Module,
+	receiverModule modules.Module,
+	parserModule modules.Module,
+	storageModule modules.Module,
+) error {
+	if err := stopperModule.AttachTo(receiverModule, receiver.StopOutput, stopper.InputName); err != nil {
+		return errors.Wrap(err, "while attaching stopper to receiver")
+	}
+
+	if err := stopperModule.AttachTo(parserModule, parser.StopOutput, stopper.InputName); err != nil {
+		return errors.Wrap(err, "while attaching stopper to parser")
+	}
+
+	if err := stopperModule.AttachTo(storageModule, storage.StopOutput, stopper.InputName); err != nil {
+		return errors.Wrap(err, "while attaching stopper to storage")
+	}
+
+	return nil
+}
diff --git a/pkg/tvl/receiver/receiver.go b/pkg/tvl/receiver/receiver.go
new file mode 100644
index 00000000..9e24c86a
--- /dev/null
+++ b/pkg/tvl/receiver/receiver.go
@@ -0,0 +1,46 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
+package receiver
+
+import (
+	"context"
+	"github.com/celenium-io/celestia-indexer/internal/tvl/l2beat"
+	"github.com/celenium-io/celestia-indexer/internal/tvl/lama"
+	"github.com/dipdup-net/indexer-sdk/pkg/modules"
+)
+
+const (
+	DataOutput = "data"
+	StopOutput = "stop"
+)
+
+type Module struct {
+	modules.BaseModule
+	apiL2Beat l2beat.IApi
+	apiLama   lama.IApi
+}
+
+func NewModule(l2beatApi l2beat.API, lamaApi lama.API) Module {
+	receiver := Module{
+		BaseModule: modules.New("receiver"),
+		apiL2Beat:  l2beatApi,
+		apiLama:    lamaApi,
+	}
+
+	receiver.CreateOutput(DataOutput)
+	receiver.CreateOutput(StopOutput)
+
+	return receiver
+}
+
+func (r *Module) Start(ctx context.Context) {
+	r.Log.Info().Msg("starting TVL receiver...")
+	//r.G.GoCtx(ctx, r.receive)
+}
+
+func (r *Module) stopAll() {
+	r.MustOutput(StopOutput).Push(struct{}{})
+}
+
+var _ modules.Module = (*Module)(nil)