AWS storage. Create image thumbnails

Author: aopoltorzhicky

.gitignore

@@ -17,5 +17,5 @@ vendor/
 *.csv
 .env*
 *.db
-config.yaml
+dipdup.yml
 build/metadata

.vscode/launch.json

@@ -10,9 +10,10 @@
             "request": "launch",
             "mode": "debug",
             "program": "${workspaceFolder}/cmd/metadata/",
+            "envFile": "${workspaceFolder}/.env",
             "args": [
-                "-f",
-                "${workspaceFolder}/config.yaml"
+                "-c",
+                "${workspaceFolder}/build/dipdup.yml"
             ]
         }
     ]

Makefile

@@ -1,3 +1,6 @@
+-include .env
+export $(shell sed 's/=.*//' .env)
+
 .PHONY: build
 
 build:
@@ -8,4 +11,8 @@ debug: build
 	cd build && POSTGRES_HOST=localhost ./metadata
 
 run:
-	docker-compose up -d --build
+	docker-compose up -d --build
+
+metadata:
+	docker-compose up -d db
+	cd cmd/metadata && go run .

build/dipdup.yml

@@ -9,6 +9,11 @@ metadata:
     ipfs_timeout: 10
     http_timeout: 10
     max_retry_count_on_error: 3
+    aws:
+      bucket_name: dipdup-thumbnails
+      region: eu-central-1
+      access_key_id: ${AWS_ACCESS_KEY_ID}
+      secret_access_key: ${AWS_SECRET_ACCESS_KEY}
   indexers:
     mainnet:
       datasources:

cmd/metadata/config/config.go

@@ -113,6 +113,7 @@ type Settings struct {
 	HTTPTimeout          uint64   `yaml:"http_timeout"`
 	MaxRetryCountOnError uint64   `yaml:"max_retry_count_on_error"`
 	Index                []string `yaml:"index"`
+	AWS                  AWS      `yaml:"aws"`
 }
 
 // Validate -
@@ -142,3 +143,11 @@ func (cfg *Settings) Validate() error {
 
 	return nil
 }
+
+// AWS -
+type AWS struct {
+	BucketName string `yaml:"bucket_name"`
+	Region     string `yaml:"region"`
+	AccessKey  string `yaml:"access_key_id"`
+	Secret     string `yaml:"secret_access_key"`
+}

cmd/metadata/contract.go

@@ -13,6 +13,9 @@ import (
 )
 
 func (indexer *Indexer) processContractMetadata(update api.BigMapUpdate, tx *gorm.DB) error {
+	if update.Content != nil {
+		return nil
+	}
 	if update.Content.Hash != emptyHash {
 		return indexer.ctx.Add(update, indexer.network)
 	}
@@ -51,7 +54,7 @@ func (indexer *Indexer) resolveContractMetadata(cm *models.ContractMetadata) {
 		switch {
 		case errors.Is(err, resolver.ErrNoIPFSResponse) || errors.Is(err, resolver.ErrTezosStorageKeyNotFound):
 			cm.RetryCount += 1
-			if cm.RetryCount < int(indexer.maxRetryCount) {
+			if cm.RetryCount < int(indexer.settings.MaxRetryCountOnError) {
 				indexer.logContractMetadata(*cm, fmt.Sprintf("Retry: %s", err.Error()), "warn")
 			} else {
 				cm.Status = models.StatusFailed

cmd/metadata/helpers/ipfs.go

@@ -0,0 +1,46 @@
+package helpers
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/ipfs/go-cid"
+)
+
+const (
+	prefixIpfs = "ipfs://"
+)
+
+// IPFSHash - separate IPFS hash from link
+func IPFSHash(link string) (string, error) {
+	hash := strings.TrimPrefix(link, prefixIpfs)
+	if _, err := cid.Decode(hash); err != nil {
+		return "", err
+	}
+	return hash, nil
+}
+
+// IPFSLink - get gateway link
+func IPFSLink(gateway, hash string) string {
+	return fmt.Sprintf("%s/ipfs/%s", gateway, hash)
+}
+
+var ipfsURL = regexp.MustCompile(`ipfs:\/\/(?P<hash>Qm[123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz]{44})`)
+
+// FindAllIPFSLinks -
+func FindAllIPFSLinks(data []byte) []string {
+	matches := ipfsURL.FindAllSubmatch(data, -1)
+	if len(matches) == 0 {
+		return nil
+	}
+
+	res := make([]string, 0)
+	for i := range matches {
+		if len(matches[i]) != 2 {
+			continue
+		}
+		res = append(res, string(matches[i][1]))
+	}
+	return res
+}

cmd/metadata/helpers/string.go

@@ -1,6 +1,7 @@
 package helpers
 
 import (
+	"bytes"
 	"encoding/hex"
 	"strings"
 )
@@ -19,3 +20,8 @@ func IsJSON(val string) bool {
 func Decode(data []byte) ([]byte, error) {
 	return hex.DecodeString(Trim(string(data)))
 }
+
+// Escape -
+func Escape(data []byte) []byte {
+	return bytes.ReplaceAll(data, []byte{0x5c, 0x75, 0x30, 0x30, 0x30, 0x30}, []byte("0x00")) // \u0000 => 0x00
+}

cmd/metadata/indexer.go

@@ -1,10 +1,10 @@
 package main
 
 import (
-	"errors"
 	"strings"
 	"sync"
 
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"gorm.io/gorm"
 
@@ -14,21 +14,23 @@ import (
 	"github.com/dipdup-net/metadata/cmd/metadata/context"
 	"github.com/dipdup-net/metadata/cmd/metadata/models"
 	"github.com/dipdup-net/metadata/cmd/metadata/resolver"
+	"github.com/dipdup-net/metadata/cmd/metadata/storage"
 	"github.com/dipdup-net/metadata/cmd/metadata/tzkt"
 )
 
 // Indexer -
 type Indexer struct {
-	network       string
-	indexName     string
-	state         state.State
-	resolver      resolver.Receiver
-	db            *gorm.DB
-	scanner       *tzkt.Scanner
-	ctx           *context.Context
-	contracts     *Queue
-	tokens        *Queue
-	maxRetryCount uint64
+	network          string
+	indexName        string
+	state            state.State
+	resolver         resolver.Receiver
+	db               *gorm.DB
+	scanner          *tzkt.Scanner
+	ctx              *context.Context
+	contracts        *Queue
+	tokens           *Queue
+	thumbnailCreator *ThumbnailCreator
+	settings         config.Settings
 
 	stop chan struct{}
 	wg   sync.WaitGroup
@@ -50,16 +52,19 @@ func NewIndexer(network string, indexerConfig *config.Indexer, database generalC
 	}
 
 	indexer := &Indexer{
-		scanner:       tzkt.New(indexerConfig.DataSource.Tzkt, filters.Accounts...),
-		network:       network,
-		indexName:     models.IndexName(network),
-		resolver:      rslvr,
-		maxRetryCount: settings.MaxRetryCountOnError,
-		ctx:           ctx,
-		db:            db,
-		stop:          make(chan struct{}, 1),
+		scanner:   tzkt.New(indexerConfig.DataSource.Tzkt, filters.Accounts...),
+		network:   network,
+		indexName: models.IndexName(network),
+		resolver:  rslvr,
+		settings:  settings,
+		ctx:       ctx,
+		db:        db,
+		stop:      make(chan struct{}, 1),
 	}
 
+	if aws := storage.NewAWS(settings.AWS.AccessKey, settings.AWS.Secret, settings.AWS.Region, settings.AWS.BucketName); aws != nil {
+		indexer.thumbnailCreator = NewThumbnailCreator(aws, db, settings.IPFSGateways)
+	}
 	indexer.contracts = NewQueue(db, 15, 60, indexer.onContractFlush, indexer.onContractTick)
 	indexer.tokens = NewQueue(db, 15, 60, indexer.onTokenFlush, indexer.onTokenTick)
 
@@ -76,6 +81,9 @@ func (indexer *Indexer) Start() error {
 		return err
 	}
 
+	if indexer.thumbnailCreator != nil {
+		indexer.thumbnailCreator.Start()
+	}
 	indexer.contracts.Start()
 	indexer.tokens.Start()
 
@@ -92,6 +100,12 @@ func (indexer *Indexer) Close() error {
 	indexer.stop <- struct{}{}
 	indexer.wg.Wait()
 
+	if indexer.thumbnailCreator != nil {
+		if err := indexer.thumbnailCreator.Close(); err != nil {
+			return err
+		}
+	}
+
 	if err := indexer.scanner.Close(); err != nil {
 		return err
 	}
@@ -175,11 +189,11 @@ func (indexer *Indexer) handlerUpdate(msg tzkt.Message) error {
 			switch path[len(path)-1] {
 			case "token_metadata":
 				if err := indexer.processTokenMetadata(msg.Body[i], tx); err != nil {
-					return err
+					return errors.Wrap(err, "token_metadata")
 				}
 			case "metadata":
 				if err := indexer.processContractMetadata(msg.Body[i], tx); err != nil {
-					return err
+					return errors.Wrap(err, "contract_metadata")
 				}
 			}
 		}

cmd/metadata/models/contract_metadata.go

@@ -17,12 +17,12 @@ type ContractMetadata struct {
 }
 
 // Status - metadata status
-type Status string
+type Status int
 
 const (
-	StatusNew     Status = "new"
-	StatusFailed  Status = "failed"
-	StatusApplied Status = "applied"
+	StatusNew Status = iota + 1
+	StatusFailed
+	StatusApplied
 )
 
 // GetContractMetadata -

cmd/metadata/models/token_metadata.go

@@ -7,14 +7,16 @@ import (
 
 // TokenMetadata -
 type TokenMetadata struct {
-	Network    string `gorm:"primaryKey"`
-	Contract   string `gorm:"primaryKey"`
-	TokenID    uint64 `gorm:"primaryKey"`
-	Link       string
-	RetryCount int
-	Status     Status
-	UpdatedAt  int
-	Metadata   datatypes.JSON
+	Network         string `gorm:"primaryKey"`
+	Contract        string `gorm:"primaryKey"`
+	TokenID         uint64 `gorm:"primaryKey"`
+	Link            string
+	RetryCount      int `gorm:"default:0"`
+	Status          Status
+	UpdatedAt       int
+	Metadata        datatypes.JSON
+	ImageProcessed  bool
+	ImageRetryCount int `gorm:"default:0"`
 }
 
 // GetTokenMetadata -
@@ -29,3 +31,22 @@ func GetTokenMetadata(tx *gorm.DB, status Status, limit, offset int) (all []Toke
 	err = query.Order("retry_count asc").Find(&all).Error
 	return
 }
+
+// GetTokenMetadataWithUnprocessedImages -
+func GetTokenMetadataWithUnprocessedImages(tx *gorm.DB) (all []TokenMetadata, err error) {
+	err = tx.Model(&TokenMetadata{}).Where("status = 3 AND image_processed = false AND image_retry_count < 3").Limit(5).Order("image_retry_count desc").Find(&all).Error
+	return
+}
+
+// SetImageProcessed -
+func (tm *TokenMetadata) SetImageProcessed(tx *gorm.DB) error {
+	tm.ImageRetryCount += 1
+	updates := map[string]interface{}{
+		"image_retry_count": tm.ImageRetryCount,
+	}
+
+	if tm.ImageProcessed {
+		updates["image_processed"] = true
+	}
+	return tx.Model(&tm).Updates(updates).Error
+}

cmd/metadata/resolver/http.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/dipdup-net/metadata/cmd/metadata/helpers"
 	"github.com/pkg/errors"
 )
 
@@ -69,7 +70,12 @@ func (s Http) Resolve(network, address, link string) ([]byte, error) {
 		return nil, errors.Errorf("Invalid status code: %s", resp.Status)
 	}
 
-	return ioutil.ReadAll(io.LimitReader(resp.Body, 20971520)) // 20 MB limit for metadata
+	data, err := ioutil.ReadAll(io.LimitReader(resp.Body, 20971520)) // 20 MB limit for metadata
+	if err != nil {
+		return nil, err
+	}
+
+	return helpers.Escape(data), nil
 }
 
 // Is -