Merge pull request #1102 from oasisprotocol/ptrus/feature/evm-neby-prices

runtime/evm_tokens: Fetch Neby price for tokens

Author: ptrus

.changelog/1102.feature.md

@@ -0,0 +1 @@
+runtime/evm_tokens: Fetch Neby price for tokens

.github/workflows/ci-test.yaml

@@ -135,7 +135,7 @@ jobs:
 
   test-e2e-regression:
     env:
-      E2E_REGRESSION_ARTIFACTS_VERSION: 2025-05-30
+      E2E_REGRESSION_ARTIFACTS_VERSION: 2025-07-21-2
     strategy:
       matrix:
         suite:
@@ -174,12 +174,13 @@ jobs:
         run: |
           make postgres
 
-      - name: Block access to Sourcify
-        # Sourcify should not need to be available in CI; we run everything from caches.
+      - name: Block access to Sourcify and Neby Graph API
+        # These services should not be available in CI; we run everything from caches.
         # Enforce unavailability to keep tests confidently reproducible, and to catch
-        # any bugs that might cause CI to accidentally use Sourcify.
+        # any bugs that might cause CI to accidentally use live external services.
         run: |
           sudo bash -c 'echo 0.0.0.0 sourcify.dev | tee -a /etc/hosts'
+          sudo bash -c 'echo 0.0.0.0 graph.api.neby.exchange | tee -a /etc/hosts'
 
       - name: Run e2e regression tests
         run : |

.golangci.yaml

@@ -31,6 +31,7 @@ linters:
             - github.com/oasisprotocol/oasis-sdk
             - github.com/oasisprotocol/metadata-registry-tools
             - github.com/akrylysov/pogreb
+            - github.com/cockroachdb/apd
             - github.com/dgraph-io/ristretto
             - github.com/ethereum/go-ethereum
             - github.com/go-chi/chi/v5

Makefile

@@ -65,7 +65,7 @@ E2E_REGRESSION_SUITES_NO_LINKS := eden_testnet_2025 eden_2025 eden damask
 # make E2E_REGRESSION_SUITES='suite1 suite2' test-e2e-regression
 E2E_REGRESSION_SUITES := $(E2E_REGRESSION_SUITES_NO_LINKS) edenfast
 
-E2E_REGRESSION_ARTIFACTS_VERSION = 2025-05-30
+E2E_REGRESSION_ARTIFACTS_VERSION = 2025-07-21-2
 
 upload-e2e-regression-caches:
 	for suite in $(E2E_REGRESSION_SUITES); do \

analyzer/neby_prices/neby_prices.go

@@ -0,0 +1,155 @@
+// Package nebyprices implements the Neby prices analyzer.
+package nebyprices
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	ethCommon "github.com/ethereum/go-ethereum/common"
+
+	"github.com/oasisprotocol/nexus/analyzer"
+	"github.com/oasisprotocol/nexus/analyzer/httpmisc"
+	"github.com/oasisprotocol/nexus/analyzer/item"
+	"github.com/oasisprotocol/nexus/common"
+	"github.com/oasisprotocol/nexus/config"
+	"github.com/oasisprotocol/nexus/log"
+	"github.com/oasisprotocol/nexus/storage"
+)
+
+const (
+	analyzerPrefix = "neby_prices_"
+
+	defaultNebyEndpoint = "https://graph.api.neby.exchange/dex"
+)
+
+type processor struct {
+	runtime common.Runtime
+	target  storage.TargetStorage
+	logger  *log.Logger
+
+	client *http.Client
+
+	endpoint string
+}
+
+var _ item.ItemProcessor[struct{}] = (*processor)(nil)
+
+func NewAnalyzer(
+	runtime common.Runtime,
+	cfg *config.NebyPricesConfig,
+	target storage.TargetStorage,
+	logger *log.Logger,
+) (analyzer.Analyzer, error) {
+	logger = logger.With("analyzer", analyzerPrefix+runtime)
+
+	logger.Info("Starting analyzer")
+
+	endpoint := cfg.Endpoint
+	if endpoint == "" {
+		endpoint = defaultNebyEndpoint
+	}
+
+	if cfg.Interval == 0 {
+		cfg.Interval = 15 * time.Minute
+	}
+	p := &processor{
+		runtime:  runtime,
+		target:   target,
+		logger:   logger,
+		client:   &http.Client{Timeout: 10 * time.Second},
+		endpoint: endpoint,
+	}
+
+	return item.NewAnalyzer(
+		analyzerPrefix+string(runtime),
+		cfg.ItemBasedAnalyzerConfig,
+		p,
+		target,
+		logger,
+	)
+}
+
+func (p *processor) GetItems(ctx context.Context, limit uint64) ([]struct{}, error) {
+	return []struct{}{{}}, nil
+}
+
+func (p *processor) ProcessItem(ctx context.Context, batch *storage.QueryBatch, item struct{}) error {
+	p.logger.Debug("fetching Neby token prices", "endpoint", p.endpoint)
+	result, err := p.fetchNebyTokenPrices(ctx, p.endpoint)
+	if err != nil {
+		return fmt.Errorf("failed to fetch Neby token prices: %w", err)
+	}
+	p.logger.Debug("fetched Neby token prices", "count", len(result.Data.Tokens), "result", result)
+
+	// Store the prices in the database.
+	for _, token := range result.Data.Tokens {
+		ethAddress := ethCommon.HexToAddress(token.ID)
+		batch.Queue(
+			tokenNebyDerivedPriceUpsert,
+			p.runtime,
+			ethAddress[:],
+			token.DerivedETH,
+		)
+	}
+
+	return nil
+}
+
+type graphTokensResponse struct {
+	Data struct {
+		Tokens []struct {
+			ID         string `json:"id"`
+			DerivedETH string `json:"derivedETH"`
+			Name       string `json:"name"`
+		} `json:"tokens"`
+	} `json:"data"`
+}
+
+func (p *processor) fetchNebyTokenPrices(ctx context.Context, endpoint string) (*graphTokensResponse, error) {
+	query := `
+	{
+		tokens(first: 1000) {
+			id
+			derivedETH
+			name
+		}
+	}`
+	body, _ := json.Marshal(map[string]string{
+		"query": query,
+	})
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := p.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to send request: %w", err)
+	}
+	if err := httpmisc.ResponseOK(resp); err != nil {
+		return nil, fmt.Errorf("response error: %w", err)
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read response body: %w", err)
+	}
+
+	var result graphTokensResponse
+	if err := json.Unmarshal(respBody, &result); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal response: %w", err)
+	}
+
+	return &result, nil
+}
+
+func (p *processor) QueueLength(ctx context.Context) (int, error) {
+	return 0, nil
+}

analyzer/neby_prices/queries.go

@@ -0,0 +1,12 @@
+package nebyprices
+
+// nolint: gosec // G101: Potential hardcoded credentials.
+const tokenNebyDerivedPriceUpsert = `
+INSERT INTO chain.evm_tokens (runtime, token_address, neby_derived_price)
+	SELECT
+		$1::runtime, derive_oasis_addr($2), $3
+	WHERE
+		derive_oasis_addr($2) IS NOT NULL
+	ON CONFLICT (runtime, token_address) DO UPDATE
+	SET
+	neby_derived_price = excluded.neby_derived_price`

analyzer/pubclient/pub_client.go

@@ -1,6 +1,7 @@
 package pubclient
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"net"
@@ -91,3 +92,12 @@ var client = &http.Client{
 func GetWithContext(ctx context.Context, url string) (*http.Response, error) {
 	return httpmisc.GetWithContextWithClient(ctx, client, url)
 }
+
+func PostWithContext(ctx context.Context, url string, bodyType string, body []byte) (*http.Response, error) {
+	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", bodyType)
+	return client.Do(req)
+}

api/spec/v1.yaml

@@ -1669,6 +1669,15 @@ components:
       example: "1234567890123456789012"
       x-go-type: common.BigInt
       x-go-type-import: { name: common, path: "github.com/oasisprotocol/nexus/common" }
+    TextBigDecimal:
+      type: string
+      pattern: '^-?[0-9]+(\.[0-9]+)?$'
+      format: decimal  # Informative only; not used by Go codegen
+      example: "12345.678901234567890123"
+      x-go-type: common.BigDecimal
+      x-go-type-import:
+        name: common
+        path: github.com/oasisprotocol/nexus/common
     Address:
       type: string
       pattern: '^oasis1[a-z0-9]{40}$'
@@ -3646,6 +3655,15 @@ components:
         total_supply:
           allOf: [$ref: '#/components/schemas/TextBigInt']
           description: The total number of base units available.
+        neby_derived_price:
+          allOf: [$ref: '#/components/schemas/TextBigDecimal']
+          description: |
+            The estimated price of one base unit of this token, expressed in the native denomination of the runtime.
+
+            This value is sourced from the Neby GraphQL API and reflects the token's `derivedETH` price as computed by the subgraph indexing the Neby DEX.
+            It is available only for tokens listed and tradable on the Neby exchange.
+
+            May be zero if the subgraph could not determine a valid price route to the native token unit.
         num_transfers:
           type: integer
           format: int64

cmd/analyzer/analyzer.go

@@ -28,6 +28,7 @@ import (
 	"github.com/oasisprotocol/nexus/analyzer/evmtokens"
 	"github.com/oasisprotocol/nexus/analyzer/evmverifier"
 	"github.com/oasisprotocol/nexus/analyzer/metadata_registry"
+	nebyprices "github.com/oasisprotocol/nexus/analyzer/neby_prices"
 	nodestats "github.com/oasisprotocol/nexus/analyzer/node_stats"
 	"github.com/oasisprotocol/nexus/analyzer/rofl"
 	roflinstance "github.com/oasisprotocol/nexus/analyzer/rofl/instance_transactions"
@@ -619,6 +620,11 @@ func NewService(ctx context.Context, cfg *config.AnalysisConfig, logger *log.Log
 			return evmabibackfill.NewAnalyzer(common.RuntimePontusxDev, cfg.Analyzers.PontusxDevAbi.ItemBasedAnalyzerConfig, dbClient, logger)
 		})
 	}
+	if cfg.Analyzers.SapphireNebyPrices != nil {
+		analyzers, err = addAnalyzer(analyzers, err, syncTagSapphire, func() (A, error) {
+			return nebyprices.NewAnalyzer(common.RuntimeSapphire, cfg.Analyzers.SapphireNebyPrices, dbClient, logger)
+		})
+	}
 	if cfg.Analyzers.MetadataRegistry != nil {
 		analyzers, err = addAnalyzer(analyzers, err, "" /*syncTag*/, func() (A, error) {
 			return metadata_registry.NewAnalyzer(*cfg.Analyzers.MetadataRegistry, dbClient, logger)

common/types.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"math/big"
 
+	"github.com/cockroachdb/apd"
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/oasisprotocol/oasis-core/go/common/quantity"
 	staking "github.com/oasisprotocol/oasis-core/go/staking/api"
@@ -172,6 +173,71 @@ func NumericToBigInt(n pgtype.Numeric) (BigInt, error) {
 	return BigInt{Int: *big0}, nil
 }
 
+// Decimal is a wrapper around apd.Decimal to allow for custom JSON marshaling.
+type BigDecimal struct {
+	apd.Decimal
+}
+
+func NewBigDecimal(v string) (BigDecimal, error) {
+	var d apd.Decimal
+	if err := d.UnmarshalText([]byte(v)); err != nil {
+		return BigDecimal{}, fmt.Errorf("invalid decimal string %q: %w", v, err)
+	}
+	return BigDecimal{d}, nil
+}
+
+func (b BigDecimal) MarshalJSON() ([]byte, error) {
+	t, err := b.MarshalText()
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal(string(t))
+}
+
+func (b *BigDecimal) UnmarshalJSON(text []byte) error {
+	var s string
+	err := json.Unmarshal(text, &s)
+	if err != nil {
+		return err
+	}
+	return b.UnmarshalText([]byte(s))
+}
+
+func (b BigDecimal) String() string {
+	return b.Decimal.String()
+}
+
+// Implement NumericValuer interface for BigInt.
+func (b BigDecimal) NumericValue() (pgtype.Numeric, error) {
+	if b.Sign() == -1 {
+		return pgtype.Numeric{}, fmt.Errorf("cannot convert negative decimal %s to Numeric", b.String())
+	}
+	return pgtype.Numeric{Int: &b.Coeff, Exp: b.Exponent, NaN: false, Valid: true, InfinityModifier: pgtype.Finite}, nil
+}
+
+// Implement NumericDecoder interface for BigInt.
+func (b *BigDecimal) ScanNumeric(n pgtype.Numeric) error {
+	if !n.Valid {
+		return fmt.Errorf("NULL values can't be decoded. Scan into a **BigDecimal to handle NULLs")
+	}
+	if n.NaN {
+		return fmt.Errorf("cannot scan NaN into *decimal.Decimal")
+	}
+
+	if n.InfinityModifier != pgtype.Finite {
+		return fmt.Errorf("cannot scan %v into *decimal.Decimal", n.InfinityModifier)
+	}
+
+	b.Decimal = apd.Decimal{
+		Coeff:    *n.Int,
+		Exponent: n.Exp,
+		Negative: false,
+		Form:     apd.Finite,
+	}
+
+	return nil
+}
+
 func Ptr[T any](v T) *T {
 	return &v
 }