Automatic Fungible Token Initialization

.env.example

@@ -30,6 +30,8 @@ FLOW_WALLET_ENABLED_TOKENS=FUSD:0xf8d6e0586b0a20c7:fusd,FlowToken:0x0ae53cb6e3f4
 # This sets the number of proposal keys to be used on the admin account.
 # You can increase transaction throughput by using multiple proposal keys for
 # parallel transaction execution.
+# WARNING: Increasing admin account's key count by more than 100 in a single transaction fails
+# due to the large event size of the resulting transaction. Please increase key count in steps of 100.
 FLOW_WALLET_ADMIN_PROPOSAL_KEY_COUNT=50
 
 # Sets the server request timeout
@@ -42,7 +44,14 @@ FLOW_WALLET_ADMIN_PROPOSAL_KEY_COUNT=50
 # Number of concurrent workers handling incoming jobs.
 # You can increase the number of workers if you're sending
 # too many transactions and find that the queue is often backlogged.
-# FLOW_WALLET_WORKER_COUNT=100 (default)
+# FLOW_WALLET_WORKER_COUNT=1 (default)
 
 # Max transactions per second, rate at which the service can submit transactions to Flow
 # FLOW_WALLET_MAX_TPS=10 (default)
+
+
+# Init enabled fungible tokens on new account creation
+INIT_FUNGIBLE_VAULTS_ON_ACCOUNT_CREATION=true
+
+# Service log level (Default=info)
+# FLOW_WALLET_LOG_LEVEL=info

README.md

@@ -108,16 +108,25 @@ When making `sync` requests it's sometimes required to adjust the server's reque
 
 ### Enabled fungible tokens
 
-A comma separated list of _fungible tokens_ and their corresponding addresses enabled for this instance. Make sure to name each token exactly as it is in the corresponding cadence code (FlowToken, FUSD etc.). Include at least FlowToken as functionality without it is undetermined.
+A comma separated list of _fungible tokens_ and their corresponding addresses and paths enabled for this instance. Make sure to name each token exactly as it is in the corresponding cadence code (FlowToken, FUSD, etc). Include at least FlowToken as functionality without it is undetermined. Format is comma separated list of:
 
-**NOTE:** It is necessary to add a 3rd parameter "lowercamelcase" name for each token. For FlowToken this would be "flowToken" and for FUSD "fusd". This is used to construct the vault name, receiver name and balance name in generic transaction templates. Consult the contract code for each token to derive the proper name (search for `.*Vault`, `.*Receiver`, `.*Balance`)
+```
+TokenName:ContractAddress:ReceiverPublicPath:BalancePublicPath:VaultStoragePath
+```
 
-**NOTE:** Non-fungible tokens can _not_ be enabled using environment variables. Use the API endpoints for that.
+Example (mainnet):
+```
+FiatToken:0xb19436aae4d94622:FiatToken.VaultReceiverPubPath:FiatToken.VaultBalancePubPath:FiatToken.VaultStoragePath
+```
+
+**DEPRECATION NOTICE:** You can optionally config each token with 3 parameters: a 3rd parameter "lowercamelcase" name for each token. For FlowToken this would be "flowToken" and for FUSD "fusd". This is used to construct the vault name, receiver name and balance name in generic transaction templates. Consult the contract code for each token to derive the proper name (search for `.*Vault`, `.*Receiver`, `.*Balance`).**THIS IS NOW DEPRECATED** It's best to grab paths from the token contract and set them explicitly here instead of generating them based on lowercase token name. The old format still works to maintain backward compatibility.
 
 Examples:
 
     FLOW_WALLET_ENABLED_TOKENS=FlowToken:0x0ae53cb6e3f42a79:flowToken,FUSD:0xf8d6e0586b0a20c7:fusd
 
+**NOTE:** Non-fungible tokens can _not_ be enabled using environment variables. Use the API endpoints for that.
+
 ### Database
 
 | Config variable | Environment variable        | Description                                                                                      | Default     | Examples                  |

accounts/account_added.go

@@ -1,12 +1,14 @@
 package accounts
 
 import (
+	"github.com/flow-hydraulics/flow-wallet-api/templates"
 	"github.com/onflow/flow-go-sdk"
 	log "github.com/sirupsen/logrus"
 )
 
 type AccountAddedPayload struct {
-	Address flow.Address
+	Address                   flow.Address
+	InitializedFungibleTokens []templates.Token
 }
 
 type accountAddedHandler interface {

accounts/service.go

@@ -13,9 +13,11 @@ import (
 	"github.com/flow-hydraulics/flow-wallet-api/flow_helpers"
 	"github.com/flow-hydraulics/flow-wallet-api/jobs"
 	"github.com/flow-hydraulics/flow-wallet-api/keys"
+	"github.com/flow-hydraulics/flow-wallet-api/templates"
 	"github.com/flow-hydraulics/flow-wallet-api/templates/template_strings"
 	"github.com/flow-hydraulics/flow-wallet-api/transactions"
 	"github.com/onflow/cadence"
+	jsoncdc "github.com/onflow/cadence/encoding/json"
 	"github.com/onflow/flow-go-sdk"
 	flow_crypto "github.com/onflow/flow-go-sdk/crypto"
 	flow_templates "github.com/onflow/flow-go-sdk/templates"
@@ -43,6 +45,7 @@ type ServiceImpl struct {
 	fc            flow_helpers.FlowClient
 	wp            jobs.WorkerPool
 	txs           transactions.Service
+	temps         templates.Service
 	txRateLimiter ratelimit.Limiter
 }
 
@@ -54,12 +57,13 @@ func NewService(
 	fc flow_helpers.FlowClient,
 	wp jobs.WorkerPool,
 	txs transactions.Service,
+	temps templates.Service,
 	opts ...ServiceOption,
 ) Service {
 	var defaultTxRatelimiter = ratelimit.NewUnlimited()
 
 	// TODO(latenssi): safeguard against nil config?
-	svc := &ServiceImpl{cfg, store, km, fc, wp, txs, defaultTxRatelimiter}
+	svc := &ServiceImpl{cfg, store, km, fc, wp, txs, temps, defaultTxRatelimiter}
 
 	for _, opt := range opts {
 		opt(svc)
@@ -359,13 +363,55 @@ func (s *ServiceImpl) createAccount(ctx context.Context) (*Account, string, erro
 		publicKeys = append(publicKeys, &clonedAccountKey)
 	}
 
-	flowTx, err := flow_templates.CreateAccount(
-		publicKeys,
-		nil,
-		payer.Address,
-	)
-	if err != nil {
-		return nil, "", err
+	var flowTx *flow.Transaction
+	var initializedFungibleTokens []templates.Token
+	if s.cfg.InitFungibleVaultsOnAccountCreation {
+
+		// Create custom cadence script to create account and init enabled fungible tokens vaults
+		tokens, err := s.temps.ListTokensFull(templates.FT)
+		if err != nil {
+			return nil, "", err
+		}
+
+		tokensInfo := []template_strings.FungibleTokenInfo{}
+		for _, t := range tokens {
+			if t.Name != "FlowToken" {
+				tokensInfo = append(tokensInfo, templates.NewFungibleTokenInfo(t))
+				initializedFungibleTokens = append(initializedFungibleTokens, t)
+			}
+		}
+
+		txScript, err := templates.CreateAccountAndInitFungibleTokenVaultsCode(s.cfg.ChainID, tokensInfo)
+		if err != nil {
+			return nil, "", err
+		}
+
+		// Encode public key list
+		keyList := make([]cadence.Value, len(publicKeys))
+		for i, key := range publicKeys {
+			keyList[i], err = flow_templates.AccountKeyToCadenceCryptoKey(key)
+			if err != nil {
+				return nil, "", err
+			}
+		}
+		cadencePublicKeys := cadence.NewArray(keyList)
+
+		flowTx = flow.NewTransaction().
+			SetScript([]byte(txScript)).
+			AddAuthorizer(payer.Address).
+			AddRawArgument(jsoncdc.MustEncode(cadencePublicKeys))
+
+	} else {
+
+		flowTx, err = flow_templates.CreateAccount(
+			publicKeys,
+			nil,
+			payer.Address,
+		)
+		if err != nil {
+			return nil, "", err
+		}
+
 	}
 
 	flowTx.
@@ -441,10 +487,13 @@ func (s *ServiceImpl) createAccount(ctx context.Context) (*Account, string, erro
 	}
 
 	AccountAdded.Trigger(AccountAddedPayload{
-		Address: flow.HexToAddress(account.Address),
+		Address:                   flow.HexToAddress(account.Address),
+		InitializedFungibleTokens: initializedFungibleTokens,
 	})
 
-	log.WithFields(log.Fields{"address": account.Address}).Debug("Account created")
+	log.
+		WithFields(log.Fields{"address": account.Address, "initialized-fungible-tokens": initializedFungibleTokens}).
+		Info("Account created")
 
 	return account, flowTx.ID().String(), nil
 }

accounts/service_init.go

@@ -55,6 +55,8 @@ func (s *ServiceImpl) InitAdminAccount(ctx context.Context) error {
 				if _, err := s.km.InitAdminProposalKeys(ctx); err != nil {
 					return err
 				}
+
+				log.Info("New admin account proposal keys created successfully")
 			}
 		} else {
 			return err

chain_events/listener.go

@@ -88,9 +88,19 @@ func (l *ListenerImpl) run(ctx context.Context, start, end uint64) error {
 		if err != nil {
 			return err
 		}
+		count := 0
 		for _, b := range r {
+			count += len(b.Events)
 			events = append(events, b.Events...)
 		}
+		log.
+			WithFields(log.Fields{
+				"type":        t,
+				"startHeight": start,
+				"endHeight":   end,
+				"resultCount": count,
+			}).
+			Debug("Fetching events")
 	}
 
 	for _, event := range events {

configs/configs.go

@@ -76,8 +76,9 @@ type Config struct {
 
 	// -- Templates --
 
-	EnabledTokens           []string `env:"ENABLED_TOKENS" envSeparator:","`
-	ScriptPathCreateAccount string   `env:"SCRIPT_PATH_CREATE_ACCOUNT" envDefault:""`
+	EnabledTokens                       []string `env:"ENABLED_TOKENS" envSeparator:","`
+	ScriptPathCreateAccount             string   `env:"SCRIPT_PATH_CREATE_ACCOUNT" envDefault:""`
+	InitFungibleVaultsOnAccountCreation bool     `env:"INIT_FUNGIBLE_VAULTS_ON_ACCOUNT_CREATION" envDefault:"false"`
 
 	// -- Workerpool --
 
@@ -128,7 +129,7 @@ type Config struct {
 	// For more info: https://pkg.go.dev/time#ParseDuration
 	ChainListenerInterval time.Duration `env:"EVENTS_INTERVAL" envDefault:"10s"`
 
-	// Max transactions per second, rate at which the service can submit transactions to Flow
+	// Max transactions per second, rate at which the service can submit transactions to Flow (excluding ops)
 	TransactionMaxSendRate int `env:"MAX_TPS" envDefault:"10"`
 
 	// maxJobErrorCount is the maximum number of times a Job can be tried to
@@ -152,6 +153,12 @@ type Config struct {
 	PauseDuration time.Duration `env:"PAUSE_DURATION" envDefault:"60s"`
 
 	GrpcMaxCallRecvMsgSize int `env:"GRPC_MAX_CALL_RECV_MSG_SIZE" envDefault:"16777216"`
+
+	// -- ops ---
+	// WorkerCount for system jobs, max number of in-flight transactions
+	OpsWorkerCount uint `env:"OPS_WORKER_COUNT" envDefault:"200"`
+	// Capacity of buffered jobs queues for system jobs.
+	OpsWorkerQueueCapacity uint `env:"OPS_WORKER_QUEUE_CAPACITY" envDefault:"300000"`
 }
 
 // Parse parses environment variables and flags to a valid Config.

configs/configs_test.go

@@ -11,6 +11,7 @@ func TestParseConfig(t *testing.T) {
 	t.Setenv("FLOW_WALLET_ENCRYPTION_KEY", "encryption-key")
 	t.Setenv("FLOW_WALLET_ENCRYPTION_KEY_TYPE", "local")
 	t.Setenv("FLOW_WALLET_ACCESS_API_HOST", "access-api-host")
+	t.Setenv("FLOW_WALLET_WORKER_COUNT", "1")
 
 	cfg, err := Parse()
 

docker-compose.test-suite.yml

@@ -2,6 +2,8 @@ version: "3.9"
 services:
   db:
     image: postgres:13-alpine
+    ports:
+      - "5432:5432"
     environment:
       POSTGRES_DB: wallet_test
       POSTGRES_USER: wallet_test

docker-compose.yml

@@ -15,6 +15,8 @@ services:
       POSTGRES_PASSWORD: wallet
     networks:
       - private
+    ports:
+      - "5432:5432"
     healthcheck:
       test:
         [
@@ -27,7 +29,7 @@ services:
 
   redis:
     image: redis:6.2-alpine
-    command: redis-server /usr/local/etc/redis/redis.conf
+    command: redis-server /usr/local/etc/redis/redis.conf --loglevel warning
     volumes:
       - ./redis-config/redis.conf:/usr/local/etc/redis/redis.conf
       - ./redis-config/users.acl:/usr/local/etc/redis/users.acl
@@ -50,6 +52,8 @@ services:
       - emulator-persist:/flowdb
     env_file:
       - ./.env
+    ports:
+      - "3569:3569"
     environment:
       FLOW_SERVICEPRIVATEKEY: ${FLOW_WALLET_ADMIN_PRIVATE_KEY}
       FLOW_SERVICEKEYSIGALGO: ECDSA_P256

flow_helpers/flow_helpers.go

@@ -50,7 +50,7 @@ func WaitForSeal(ctx context.Context, flowClient FlowClient, id flow.Identifier,
 
 	b := &backoff.Backoff{
 		Min:    100 * time.Millisecond,
-		Max:    time.Minute,
+		Max:    time.Second,
 		Factor: 5,
 		Jitter: true,
 	}

handlers/ops.go

@@ -0,0 +1,27 @@
+package handlers
+
+import (
+	"net/http"
+
+	"github.com/flow-hydraulics/flow-wallet-api/ops"
+)
+
+// Ops is a HTTP server for admin (system) operations.
+type Ops struct {
+	service ops.Service
+}
+
+// NewOps initiates a new ops server.
+func NewOps(service ops.Service) *Ops {
+	return &Ops{service}
+}
+
+// InitMissingFungibleVaults starts the job to initialize missing fungible token vaults.
+func (s *Ops) InitMissingFungibleVaults() http.Handler {
+	return http.HandlerFunc(s.InitMissingFungibleVaultsFunc)
+}
+
+// GetMissingFungibleVaults returns number of accounts that are missing a configured fungible token vault.
+func (s *Ops) GetMissingFungibleVaults() http.Handler {
+	return http.HandlerFunc(s.GetMissingFungibleVaultsFunc)
+}

handlers/ops_func.go

@@ -0,0 +1,29 @@
+package handlers
+
+import (
+	"net/http"
+)
+
+// InitMissingFungibleVaultsFunc starts job to init missing fungible token vaults.
+func (s *Ops) InitMissingFungibleVaultsFunc(rw http.ResponseWriter, r *http.Request) {
+
+	result, err := s.service.InitMissingFungibleTokenVaults()
+	if err != nil {
+		handleError(rw, r, err)
+		return
+	}
+
+	handleJsonResponse(rw, http.StatusOK, result)
+}
+
+// GetMissingFungibleVaultsFunc returns number of accounts with missing fungible token vaults.
+func (s *Ops) GetMissingFungibleVaultsFunc(rw http.ResponseWriter, r *http.Request) {
+
+	result, err := s.service.GetMissingFungibleTokenVaults()
+	if err != nil {
+		handleError(rw, r, err)
+		return
+	}
+
+	handleJsonResponse(rw, http.StatusOK, result)
+}

main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/flow-hydraulics/flow-wallet-api/jobs"
 	"github.com/flow-hydraulics/flow-wallet-api/keys"
 	"github.com/flow-hydraulics/flow-wallet-api/keys/basic"
+	"github.com/flow-hydraulics/flow-wallet-api/ops"
 	"github.com/flow-hydraulics/flow-wallet-api/system"
 	"github.com/flow-hydraulics/flow-wallet-api/templates"
 	"github.com/flow-hydraulics/flow-wallet-api/tokens"
@@ -128,8 +129,9 @@ func runServer(cfg *configs.Config) {
 	templateService := templates.NewService(cfg, templates.NewGormStore(db))
 	jobsService := jobs.NewService(jobs.NewGormStore(db))
 	transactionService := transactions.NewService(cfg, transactions.NewGormStore(db), km, fc, wp, transactions.WithTxRatelimiter(txRatelimiter))
-	accountService := accounts.NewService(cfg, accounts.NewGormStore(db), km, fc, wp, transactionService, accounts.WithTxRatelimiter(txRatelimiter))
+	accountService := accounts.NewService(cfg, accounts.NewGormStore(db), km, fc, wp, transactionService, templateService, accounts.WithTxRatelimiter(txRatelimiter))
 	tokenService := tokens.NewService(cfg, tokens.NewGormStore(db), km, fc, wp, transactionService, templateService, accountService)
+	opsService := ops.NewService(cfg, ops.NewGormStore(db), templateService, transactionService, tokenService)
 
 	// Register a handler for account added events
 	accounts.AccountAdded.Register(&tokens.AccountAddedHandler{
@@ -152,6 +154,7 @@ func runServer(cfg *configs.Config) {
 	accountHandler := handlers.NewAccounts(accountService)
 	transactionHandler := handlers.NewTransactions(transactionService)
 	tokenHandler := handlers.NewTokens(tokenService)
+	opsHandler := handlers.NewOps(opsService)
 
 	r := mux.NewRouter()
 
@@ -241,6 +244,10 @@ func runServer(cfg *configs.Config) {
 		log.Info("non-fungible tokens disabled")
 	}
 
+	// Ops
+	rv.Handle("/ops/missing-fungible-token-vaults/start", opsHandler.InitMissingFungibleVaults()).Methods(http.MethodGet) // start retroactive init job
+	rv.Handle("/ops/missing-fungible-token-vaults/stats", opsHandler.GetMissingFungibleVaults()).Methods(http.MethodGet)  // get number of accounts with missing fungible token vaults
+
 	h := http.TimeoutHandler(r, cfg.ServerRequestTimeout, "request timed out")
 	h = handlers.UseCors(h)
 	h = handlers.UseLogging(h)