analyzer: add evmnfts

analyzer/evmnfts/evm_nfts.go

@@ -0,0 +1,127 @@
+package evmnfts
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+
+	"github.com/oasisprotocol/nexus/analyzer"
+	"github.com/oasisprotocol/nexus/analyzer/evmnfts/ipfsclient"
+	"github.com/oasisprotocol/nexus/analyzer/item"
+	"github.com/oasisprotocol/nexus/analyzer/queries"
+	"github.com/oasisprotocol/nexus/analyzer/runtime/evm"
+	"github.com/oasisprotocol/nexus/common"
+	"github.com/oasisprotocol/nexus/config"
+	"github.com/oasisprotocol/nexus/log"
+	"github.com/oasisprotocol/nexus/storage"
+	"github.com/oasisprotocol/nexus/storage/client"
+	"github.com/oasisprotocol/nexus/storage/oasis/nodeapi"
+)
+
+const (
+	evmNFTsAnalyzerPrefix = "evm_nfts_"
+)
+
+type processor struct {
+	runtime    common.Runtime
+	source     nodeapi.RuntimeApiLite
+	ipfsClient ipfsclient.Client
+	target     storage.TargetStorage
+	logger     *log.Logger
+}
+
+var _ item.ItemProcessor[*StaleNFT] = (*processor)(nil)
+
+func NewAnalyzer(
+	runtime common.Runtime,
+	cfg config.ItemBasedAnalyzerConfig,
+	sourceClient nodeapi.RuntimeApiLite,
+	ipfsClient ipfsclient.Client,
+	target storage.TargetStorage,
+	logger *log.Logger,
+) (analyzer.Analyzer, error) {
+	logger = logger.With("analyzer", evmNFTsAnalyzerPrefix+runtime)
+	p := &processor{
+		runtime:    runtime,
+		source:     sourceClient,
+		ipfsClient: ipfsClient,
+		target:     target,
+		logger:     logger,
+	}
+	return item.NewAnalyzer[*StaleNFT](
+		evmNFTsAnalyzerPrefix+string(runtime),
+		cfg,
+		p,
+		target,
+		logger,
+	)
+}
+
+type StaleNFT struct {
+	Addr                  string
+	ID                    *big.Int
+	Type                  *evm.EVMTokenType
+	AddrContextIdentifier string
+	AddrContextVersion    int
+	AddrData              []byte
+	DownloadRound         uint64
+}
+
+func (p *processor) GetItems(ctx context.Context, limit uint64) ([]*StaleNFT, error) {
+	var staleNFTs []*StaleNFT
+	rows, err := p.target.Query(ctx, queries.RuntimeEVMNFTAnalysisStale, p.runtime, limit)
+	if err != nil {
+		return nil, fmt.Errorf("querying discovered NFTs: %w", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var staleNFT StaleNFT
+		var idC common.BigInt
+		if err = rows.Scan(
+			&staleNFT.Addr,
+			&idC,
+			&staleNFT.Type,
+			&staleNFT.AddrContextIdentifier,
+			&staleNFT.AddrContextVersion,
+			&staleNFT.AddrData,
+			&staleNFT.DownloadRound,
+		); err != nil {
+			return nil, fmt.Errorf("scanning discovered nft: %w", err)
+		}
+		staleNFT.ID = &idC.Int
+		staleNFTs = append(staleNFTs, &staleNFT)
+	}
+	return staleNFTs, nil
+}
+
+func (p *processor) ProcessItem(ctx context.Context, batch *storage.QueryBatch, staleNFT *StaleNFT) error {
+	p.logger.Info("downloading", "stale_nft", staleNFT)
+	tokenEthAddr, err := client.EVMEthAddrFromPreimage(staleNFT.AddrContextIdentifier, staleNFT.AddrContextVersion, staleNFT.AddrData)
+	if err != nil {
+		return fmt.Errorf("token address: %w", err)
+	}
+	nftData, err := evm.EVMDownloadNewNFT(
+		ctx,
+		p.logger,
+		p.source,
+		p.ipfsClient,
+		staleNFT.DownloadRound,
+		tokenEthAddr,
+		*staleNFT.Type,
+		staleNFT.ID,
+	)
+	if err != nil {
+		return fmt.Errorf("downloading NFT %s %v: %w", staleNFT.Addr, staleNFT.ID, err)
+	}
+	batch.Queue(queries.RuntimeEVMNFTUpdate, p.runtime, staleNFT.Addr, staleNFT.ID, nftData.MetadataURI, nftData.MetadataAccessed, nftData.Name, nftData.Description, nftData.Image)
+	batch.Queue(queries.RuntimeEVMNFTAnalysisUpdate, p.runtime, staleNFT.Addr, staleNFT.ID, staleNFT.DownloadRound)
+	return nil
+}
+
+func (p *processor) QueueLength(ctx context.Context) (int, error) {
+	var queueLength int
+	if err := p.target.QueryRow(ctx, queries.RuntimeEVMNFTAnalysisStaleCount, p.runtime).Scan(&queueLength); err != nil {
+		return 0, fmt.Errorf("querying number of stale NFTs: %w", err)
+	}
+	return queueLength, nil
+}

analyzer/queries/queries.go

@@ -586,13 +586,69 @@ var (
       runtime = $1 AND
       token_address = $2`
 
+	RuntimeEVMNFTUpdate = `
+    UPDATE chain.evm_nfts SET
+      metadata_uri = $4,
+      metadata_accessed = $5,
+      name = $6,
+      description = $7,
+      image = $8
+    WHERE
+      runtime = $1 AND
+      token_address = $2 AND
+      nft_id = $3`
+
 	RuntimeEVMNFTInsert = `
     INSERT INTO chain.evm_nfts
       (runtime, token_address, nft_id, last_want_download_round)
     VALUES
       ($1, $2, $3, $4)
     ON CONFLICT (runtime, token_address, nft_id) DO NOTHING`
 
+	RuntimeEVMNFTAnalysisStale = `
+    SELECT
+      chain.evm_nfts.token_address,
+      chain.evm_nfts.nft_id,
+      chain.evm_tokens.token_type,
+      chain.address_preimages.context_identifier,
+      chain.address_preimages.context_version,
+      chain.address_preimages.address_data,
+      (
+        SELECT MAX(height)
+        FROM analysis.processed_blocks
+        WHERE
+          analysis.processed_blocks.analyzer = chain.evm_nfts.runtime::TEXT AND
+          processed_time IS NOT NULL
+      ) AS download_round
+    FROM chain.evm_nfts
+    JOIN chain.evm_tokens USING
+      (runtime, token_address)
+    LEFT JOIN chain.address_preimages ON
+      chain.address_preimages.address = chain.evm_nfts.token_address
+    WHERE
+      chain.evm_nfts.runtime = $1 AND
+      (
+          chain.evm_nfts.last_download_round IS NULL OR
+          chain.evm_nfts.last_want_download_round > chain.evm_nfts.last_download_round
+      )
+    LIMIT $2`
+
+	RuntimeEVMNFTAnalysisStaleCount = `
+    SELECT COUNT(*) AS cnt
+    FROM chain.evm_nfts
+    WHERE
+      runtime = $1 AND
+      (last_download_round IS NULL OR last_want_download_round > last_download_round)`
+
+	RuntimeEVMNFTAnalysisUpdate = `
+    UPDATE chain.evm_nfts
+    SET
+      last_download_round = $4
+    WHERE
+      runtime = $1 AND
+      token_address = $2 AND
+      nft_id = $3`
+
 	RuntimeEVMTokenBalanceAnalysisStale = fmt.Sprintf(`
     WITH
     max_processed_round AS (

analyzer/runtime/evm/client.go

@@ -5,12 +5,14 @@ import (
 	"encoding/hex"
 	"fmt"
 	"math/big"
+	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	ethCommon "github.com/ethereum/go-ethereum/common"
 	"github.com/oasisprotocol/oasis-core/go/common/cbor"
 	sdkTypes "github.com/oasisprotocol/oasis-sdk/client-sdk/go/types"
 
+	"github.com/oasisprotocol/nexus/analyzer/evmnfts/ipfsclient"
 	apiTypes "github.com/oasisprotocol/nexus/api/v1/types"
 	"github.com/oasisprotocol/nexus/common"
 	"github.com/oasisprotocol/nexus/log"
@@ -56,6 +58,14 @@ type EVMTokenMutableData struct {
 	TotalSupply *big.Int
 }
 
+type EVMNFTData struct {
+	MetadataURI      string
+	MetadataAccessed time.Time
+	Name             string
+	Description      string
+	Image            string
+}
+
 type EVMTokenBalanceData struct {
 	// Balance... if you're here to ask about why there's a "balance" struct
 	// with a Balance field, it's because the struct is really a little
@@ -242,6 +252,20 @@ func EVMDownloadMutatedToken(ctx context.Context, logger *log.Logger, source nod
 	}
 }
 
+func EVMDownloadNewNFT(ctx context.Context, logger *log.Logger, source nodeapi.RuntimeApiLite, ipfsClient ipfsclient.Client, round uint64, tokenEthAddr []byte, tokenType EVMTokenType, id *big.Int) (*EVMNFTData, error) {
+	switch tokenType {
+	case EVMTokenTypeERC721:
+		nftData, err := evmDownloadNFTERC721(ctx, logger, source, ipfsClient, round, tokenEthAddr, id)
+		if err != nil {
+			return nil, fmt.Errorf("download NFT ERC-721: %w", err)
+		}
+		return nftData, nil
+
+	default:
+		return nil, fmt.Errorf("download stale nft type %v not handled", tokenType)
+	}
+}
+
 // EVMDownloadTokenBalance tries to download the balance of a given account
 // for a given token. If it transiently fails to download the balance, it
 // returns with a non-nil error. If it deterministically cannot download the

analyzer/runtime/evm/erc721.go

@@ -3,18 +3,40 @@ package evm
 import (
 	"context"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
+	"io"
+	"math/big"
+	"time"
 
 	ethCommon "github.com/ethereum/go-ethereum/common"
 	"github.com/oasisprotocol/oasis-core/go/common/errors"
 
 	"github.com/oasisprotocol/nexus/analyzer/evmabi"
+	"github.com/oasisprotocol/nexus/analyzer/evmnfts/ipfsclient"
+	"github.com/oasisprotocol/nexus/analyzer/evmnfts/multiproto"
 	"github.com/oasisprotocol/nexus/log"
 	"github.com/oasisprotocol/nexus/storage/oasis/nodeapi"
 )
 
 const EVMTokenTypeERC721 EVMTokenType = 721
 
+const MaxMetadataBytes = 10 * 1024 * 1024
+
+// ERC721AssetMetadata is asset metadata
+// https://eips.ethereum.org/EIPS/eip-721
+type ERC721AssetMetadata struct {
+	// Name identifies the asset to which this NFT represents
+	Name string `json:"name"`
+	// Description describes the asset to which this NFT represents
+	Description string `json:"description"`
+	// Image is A URI pointing to a resource with mime type image/*
+	// representing the asset to which this NFT represents. Consider making
+	// any images at a width between 320 and 1080 pixels and aspect ratio
+	// between 1.91:1 and 4:5 inclusive.
+	Image string `json:"image"`
+}
+
 func evmDownloadTokenERC721Mutable(ctx context.Context, logger *log.Logger, source nodeapi.RuntimeApiLite, round uint64, tokenEthAddr []byte) (*EVMTokenMutableData, error) {
 	var mutable EVMTokenMutableData
 	supportsEnumerable, err := detectInterface(ctx, logger, source, round, tokenEthAddr, ERC721EnumerableInterfaceID)
@@ -62,6 +84,58 @@ func evmDownloadTokenERC721(ctx context.Context, logger *log.Logger, source node
 	return &tokenData, nil
 }
 
+func evmDownloadNFTERC721(ctx context.Context, logger *log.Logger, source nodeapi.RuntimeApiLite, ipfsClient ipfsclient.Client, round uint64, tokenEthAddr []byte, id *big.Int) (*EVMNFTData, error) {
+	var nftData EVMNFTData
+	supportsMetadata, err := detectInterface(ctx, logger, source, round, tokenEthAddr, ERC721MetadataInterfaceID)
+	if err != nil {
+		return nil, fmt.Errorf("checking ERC721Metadata interface: %w", err)
+	}
+	//nolint:nestif
+	if supportsMetadata {
+		if err = evmCallWithABI(ctx, source, round, tokenEthAddr, evmabi.ERC721Metadata, &nftData.MetadataURI, "tokenURI", id); err != nil {
+			if !errors.Is(err, EVMDeterministicError{}) {
+				return nil, fmt.Errorf("calling tokenURI: %w", err)
+			}
+			logDeterministicError(logger, round, tokenEthAddr, "ERC721Metadata", "tokenURI", err,
+				"nft_id", id,
+			)
+		}
+		if nftData.MetadataURI != "" {
+			logger.Info("downloading metadata",
+				"token_eth_addr", hex.EncodeToString(tokenEthAddr),
+				"token_id", id,
+				"uri", nftData.MetadataURI,
+			)
+			nftData.MetadataAccessed = time.Now()
+			rc, err1 := multiproto.Get(ctx, ipfsClient, nftData.MetadataURI)
+			if err1 != nil {
+				// todo: retry on some errors?
+				logger.Info("error downloading token metadata",
+					"uri", nftData.MetadataURI,
+					"err", err1,
+				)
+			}
+			if rc != nil {
+				limitedReader := io.LimitReader(rc, MaxMetadataBytes)
+				var metadata ERC721AssetMetadata
+				if err1 = json.NewDecoder(limitedReader).Decode(&metadata); err1 != nil {
+					logger.Info("error decoding token metadata",
+						"uri", nftData.MetadataURI,
+						"err", err1,
+					)
+				}
+				if err1 = rc.Close(); err1 != nil {
+					return nil, fmt.Errorf("closing metadata reader: %w", err1)
+				}
+				nftData.Name = metadata.Name
+				nftData.Description = metadata.Description
+				nftData.Image = metadata.Image
+			}
+		}
+	}
+	return &nftData, nil
+}
+
 func evmDownloadTokenBalanceERC721(ctx context.Context, logger *log.Logger, source nodeapi.RuntimeApiLite, round uint64, tokenEthAddr []byte, accountEthAddr []byte) (*EVMTokenBalanceData, error) {
 	var balanceData EVMTokenBalanceData
 	accountECAddr := ethCommon.BytesToAddress(accountEthAddr)