Rescan tokens to process missing transactions

.golangci.yml

@@ -1,10 +1,11 @@
 run:
         go: '1.20'
-        skip-files:
-         - scanner/providers/gitcoin/gitcoin_stamps.go
 issues:
         max-same-issues: 0
         exclude-use-default: false
+        exclude-files:
+                - scanner/providers/gitcoin/gitcoin_stamps.go
+                - scripts/*
 linters:
         enable:
         - misspell

api/api.go

@@ -18,6 +18,7 @@ import (
 	queries "github.com/vocdoni/census3/db/sqlc"
 	"github.com/vocdoni/census3/helpers/queue"
 	"github.com/vocdoni/census3/helpers/web3"
+	"github.com/vocdoni/census3/scanner"
 	"github.com/vocdoni/census3/scanner/providers"
 	"github.com/vocdoni/census3/scanner/providers/manager"
 	web3provider "github.com/vocdoni/census3/scanner/providers/web3"
@@ -39,8 +40,10 @@ type Census3APIConf struct {
 	Hostname        string
 	Port            int
 	DataDir         string
+	FiltersPath     string
 	GroupKey        string
 	Web3Providers   *web3.Web3Pool
+	TokenUpdater    *scanner.Updater
 	HolderProviders *manager.ProviderManager
 	AdminToken      string
 }
@@ -57,6 +60,8 @@ type census3API struct {
 	holderProviders *manager.ProviderManager
 	cache           *lru.Cache[CacheKey, any]
 	router          *httprouter.HTTProuter
+	tokenUpdater    *scanner.Updater
+	filtersPath     string
 }
 
 func Init(db *db.DB, conf Census3APIConf) (*census3API, error) {
@@ -70,8 +75,10 @@ func Init(db *db.DB, conf Census3APIConf) (*census3API, error) {
 		w3p:             conf.Web3Providers,
 		queue:           queue.NewBackgroundQueue(),
 		holderProviders: conf.HolderProviders,
+		tokenUpdater:    conf.TokenUpdater,
 		cache:           cache,
 		router:          &httprouter.HTTProuter{},
+		filtersPath:     conf.FiltersPath,
 	}
 	// get the current chainID
 	log.Infow("starting API", "web3Providers", conf.Web3Providers.String())

api/errors.go

@@ -128,6 +128,16 @@ var (
 		HTTPstatus: apirest.HTTPstatusNotFound,
 		Err:        fmt.Errorf("token holder not found for the token provided"),
 	}
+	ErrNoSyncedToken = apirest.APIerror{
+		Code:       4024,
+		HTTPstatus: apirest.HTTPstatusBadRequest,
+		Err:        fmt.Errorf("token is not synced yet"),
+	}
+	ErrMalformedRescanQueueID = apirest.APIerror{
+		Code:       4025,
+		HTTPstatus: apirest.HTTPstatusBadRequest,
+		Err:        fmt.Errorf("malformed queue ID"),
+	}
 	ErrCantCreateToken = apirest.APIerror{
 		Code:       5000,
 		HTTPstatus: apirest.HTTPstatusInternalErr,

api/tokens.go

@@ -19,11 +19,13 @@ import (
 	queries "github.com/vocdoni/census3/db/sqlc"
 	"github.com/vocdoni/census3/helpers/lexer"
 	"github.com/vocdoni/census3/metrics"
+	"github.com/vocdoni/census3/scanner"
 	"github.com/vocdoni/census3/scanner/providers"
 	"github.com/vocdoni/census3/scanner/providers/web3"
 	"go.vocdoni.io/dvote/httprouter"
 	api "go.vocdoni.io/dvote/httprouter/apirest"
 	"go.vocdoni.io/dvote/log"
+	"go.vocdoni.io/dvote/util"
 )
 
 func (capi *census3API) initTokenHandlers() error {
@@ -43,6 +45,14 @@ func (capi *census3API) initTokenHandlers() error {
 		api.MethodAccessTypePublic, capi.tokenStartBlock); err != nil {
 		return err
 	}
+	if err := capi.endpoint.RegisterMethod("/tokens/update/{tokenID}", "POST",
+		api.MethodAccessTypeAdmin, capi.rescanToken); err != nil {
+		return err
+	}
+	if err := capi.endpoint.RegisterMethod("/tokens/update/queue/{queueID}", "GET",
+		api.MethodAccessTypeAdmin, capi.checkRescanToken); err != nil {
+		return err
+	}
 	if err := capi.endpoint.RegisterMethod("/tokens/{tokenID}", "DELETE",
 		api.MethodAccessTypeAdmin, capi.launchDeleteToken); err != nil {
 		return err
@@ -607,6 +617,100 @@ func (capi *census3API) getToken(msg *api.APIdata, ctx *httprouter.HTTPContext)
 	return ctx.Send(res, api.HTTPstatusOK)
 }
 
+// rescanToken function handler enqueues the rescan process for the token with
+// the given ID. The token is scanned from the creation block to the last block
+// stored in the database. It returns a 400 error if the provided ID is wrong or
+// empty, a 404 error if the token is not found, a 500 error if something fails
+// or a 200 response if the process is enqueued. It returns a queue ID to track
+// the status of the process.
+func (capi *census3API) rescanToken(msg *api.APIdata, ctx *httprouter.HTTPContext) error {
+	// get contract address from the tokenID query param and decode check if
+	// it is provided, if not return an error
+	strAddress := ctx.URLParam("tokenID")
+	if strAddress == "" {
+		return ErrMalformedToken.With("tokenID is required")
+	}
+	address := common.HexToAddress(strAddress)
+	// get chainID from query params and decode it as integer, if it's not
+	// provided or it's not a valid integer return an error
+	strChainID := ctx.Request.URL.Query().Get("chainID")
+	if strChainID == "" {
+		return ErrMalformedChainID.With("chainID is required")
+	}
+	chainID, err := strconv.Atoi(strChainID)
+	if err != nil {
+		return ErrMalformedChainID.WithErr(err)
+	} else if chainID < 0 {
+		return ErrMalformedChainID.With("chainID must be a positive number")
+	}
+	// get token information from the database
+	internalCtx, cancel := context.WithTimeout(ctx.Request.Context(), getTokenTimeout)
+	defer cancel()
+	tokenData, err := capi.db.QueriesRO.GetToken(internalCtx,
+		queries.GetTokenParams{
+			ID:      address.Bytes(),
+			ChainID: uint64(chainID),
+		})
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFoundToken.WithErr(err)
+		}
+		return ErrCantGetToken.WithErr(err)
+	}
+	// only the tokens that are already synced can be rescanned
+	if !tokenData.Synced {
+		return ErrNoSyncedToken
+	}
+	// enqueue the rescan token process
+	id := util.RandomHex(4)
+	if err := capi.tokenUpdater.SetRequest(id, &scanner.UpdateRequest{
+		Address:       address,
+		ChainID:       uint64(chainID),
+		Type:          tokenData.TypeID,
+		CreationBlock: uint64(tokenData.CreationBlock),
+		EndBlock:      uint64(tokenData.LastBlock),
+	}); err != nil {
+		return ErrEncodeQueueItem.WithErr(err)
+	}
+	// encoding the result and response it
+	res, err := json.Marshal(QueueResponse{id})
+	if err != nil {
+		return ErrEncodeQueueItem.WithErr(err)
+	}
+	return ctx.Send(res, api.HTTPstatusOK)
+}
+
+// checkRescanToken function handler returns the status of the rescan process
+// with the given queue ID. It returns a 400 error if the provided ID is wrong
+// or empty, a 404 error if the token is not found in the queue or a 500 error
+// if something fails. The response contains the address of the token, the chain
+// ID, the status of the process, the number of logs scanned, the number of new
+// logs found, and the number of duplicated logs.
+func (capi *census3API) checkRescanToken(msg *api.APIdata, ctx *httprouter.HTTPContext) error {
+	queueID := ctx.URLParam("queueID")
+	if queueID == "" {
+		return ErrMalformedRescanQueueID
+	}
+	// get the rescan status from the updater
+	status := capi.tokenUpdater.RequestStatus(queueID, true)
+	if status == nil {
+		return ErrNotFoundToken.Withf("the ID %s does not exist in the queue", queueID)
+	}
+	// encoding the result and response it
+	response, err := json.Marshal(RescanTokenStatus{
+		Address:        status.Address.String(),
+		ChainID:        status.ChainID,
+		Done:           status.Done,
+		LogsScanned:    status.TotalLogs,
+		NewLogs:        status.TotalNewLogs,
+		DuplicatedLogs: status.TotalAlreadyProcessedLogs,
+	})
+	if err != nil {
+		return ErrEncodeQueueItem.WithErr(err)
+	}
+	return ctx.Send(response, api.HTTPstatusOK)
+}
+
 func (capi *census3API) tokenStartBlock(msg *api.APIdata, ctx *httprouter.HTTPContext) error {
 	req := Token{}
 	if err := json.Unmarshal(msg.Data, &req); err != nil {

api/types.go

@@ -162,3 +162,17 @@ type DeleteTokenQueueResponse struct {
 	Done  bool  `json:"done"`
 	Error error `json:"error"`
 }
+
+type RescanTokenResponse struct {
+	ID string `json:"ID"`
+}
+
+type RescanTokenStatus struct {
+	Address string `json:"address"`
+	ChainID uint64 `json:"chainID"`
+	Done    bool   `json:"done"`
+
+	LogsScanned    uint64 `json:"logsScanned"`
+	NewLogs        uint64 `json:"newLogs"`
+	DuplicatedLogs uint64 `json:"duplicatedLogs"`
+}

cmd/census3/main.go

@@ -24,6 +24,8 @@ import (
 	"github.com/vocdoni/census3/scanner/providers/manager"
 	"github.com/vocdoni/census3/scanner/providers/poap"
 	web3provider "github.com/vocdoni/census3/scanner/providers/web3"
+	dvotedb "go.vocdoni.io/dvote/db"
+	"go.vocdoni.io/dvote/db/metadb"
 	"go.vocdoni.io/dvote/log"
 )
 
@@ -39,6 +41,7 @@ type Census3Config struct {
 	adminToken                     string
 	initialTokens                  string
 	farcaster                      bool
+	filtersPath                    string
 }
 
 func main() {
@@ -135,6 +138,12 @@ func main() {
 		panic(err)
 	}
 	config.farcaster = pviper.GetBool("farcaster")
+	// set the filters path into the config, create the folder if it does not
+	// exitst yet
+	config.filtersPath = config.dataDir + "/filters"
+	if err := os.MkdirAll(config.filtersPath, os.ModePerm); err != nil {
+		log.Fatal(err)
+	}
 	// init logger
 	log.Init(config.logLevel, "stdout", nil)
 	// check if the web3 providers are defined
@@ -194,8 +203,15 @@ func main() {
 			DB:        farcasterDB,
 		})
 	}
+	// init the filters database
+	filtersDB, err := metadb.New(dvotedb.TypePebble, config.filtersPath)
+	if err != nil {
+		log.Fatal(err)
+	}
+	// start the token updater with the database and the provider manager
+	updater := scanner.NewUpdater(database, w3p, pm, filtersDB, config.scannerCoolDown)
 	// start the holder scanner with the database and the provider manager
-	hc := scanner.NewScanner(database, w3p, pm, config.scannerCoolDown)
+	hc := scanner.NewScanner(database, updater, w3p, pm, config.scannerCoolDown)
 	// if the admin token is not defined, generate a random one
 	if config.adminToken != "" {
 		if _, err := uuid.Parse(config.adminToken); err != nil {
@@ -216,6 +232,7 @@ func main() {
 		GroupKey:        config.connectKey,
 		HolderProviders: pm,
 		AdminToken:      config.adminToken,
+		TokenUpdater:    updater,
 	})
 	if err != nil {
 		log.Fatal(err)
@@ -228,7 +245,8 @@ func main() {
 		log.Info("initial tokens created, or at least tried to")
 	}()
 	// start the holder scanner
-	go hc.Start(ctx, config.scannerConcurrentTokens)
+	go hc.Start(ctx)
+	go updater.Start(ctx, config.scannerConcurrentTokens)
 
 	metrics.NewCounter(fmt.Sprintf("census3_info{version=%q,chains=%q}",
 		internal.Version, w3p.String())).Set(1)
@@ -243,6 +261,7 @@ func main() {
 	// closing database
 	go func() {
 		hc.Stop()
+		updater.Stop()
 		if err := apiService.Stop(); err != nil {
 			log.Fatal(err)
 		}