refactor: cmd/service/* and api

cmd/init_api.go

@@ -1,43 +0,0 @@
-package main
-
-import (
-	"net/http"
-
-	"github.com/arl/statsviz"
-	"github.com/go-playground/validator"
-	"github.com/grassrootseconomics/cic-custodial/internal/api"
-	"github.com/labstack/echo/v4"
-)
-
-func initApiServer() *echo.Echo {
-	lo.Debug("bootstrapping api server")
-	server := echo.New()
-	server.HideBanner = true
-	server.HidePort = true
-
-	if ko.Bool("service.statsviz_debug") {
-		lo.Debug("Starting stats_viz at /debug/statsviz")
-		statsVizMux := http.NewServeMux()
-		_ = statsviz.Register(statsVizMux)
-		server.GET("/debug/statsviz/", echo.WrapHandler(statsVizMux))
-		server.GET("/debug/statsviz/*", echo.WrapHandler(statsVizMux))
-	}
-
-	server.Validator = &api.CustomValidator{
-		Validator: validator.New(),
-	}
-
-	apiRoute := server.Group("/api")
-
-	apiRoute.POST("/register", api.RegistrationHandler(
-		taskerClient,
-		postgresKeystore,
-	))
-
-	apiRoute.POST("/transfer", api.TransferHandler(
-		taskerClient,
-	))
-
-	lo.Debug("Registered all api handlers")
-	return server
-}

cmd/init_system.go

@@ -1,58 +0,0 @@
-package main
-
-import (
-	"context"
-	"crypto/ecdsa"
-	"math/big"
-	"time"
-
-	eth_crypto "github.com/celo-org/celo-blockchain/crypto"
-	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
-	"github.com/grassrootseconomics/w3-celo-patch"
-)
-
-func initSystemContainer() *tasker.SystemContainer {
-	return &tasker.SystemContainer{
-		Abis:                  initAbis(),
-		GasRefillThreshold:    big.NewInt(ko.MustInt64("system.gas_refill_threshold")),
-		GasRefillValue:        big.NewInt(ko.MustInt64("system.gas_refill_value")),
-		GiftableGasValue:      big.NewInt(ko.MustInt64("system.giftable_gas_value")),
-		GiftableToken:         w3.A(ko.MustString("system.giftable_token_address")),
-		GiftableTokenValue:    big.NewInt(ko.MustInt64("system.giftable_token_value")),
-		LockPrefix:            ko.MustString("system.lock_prefix"),
-		LockTimeout:           1 * time.Second,
-		PrivateKey:            initSystemKey(),
-		PublicKey:             ko.MustString("system.public_key"),
-		TokenDecimals:         ko.MustInt("system.token_decimals"),
-		TokenTransferGasLimit: uint64(ko.MustInt64("system.token_transfer_gas_limit")),
-	}
-}
-
-func initAbis() map[string]*w3.Func {
-	return map[string]*w3.Func{
-		"mint":     w3.MustNewFunc("mint(address,uint256)", ""),
-		"transfer": w3.MustNewFunc("transfer(address,uint256)", "bool"),
-	}
-}
-
-func initSystemKey() *ecdsa.PrivateKey {
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	currentSystemNonce, err := redisNoncestore.Peek(ctx, ko.MustString("system.public_key"))
-	lo.Debug("initNoncestore: loaded (noncestore) system nonce", "nonce", currentSystemNonce)
-	if err != nil {
-		nonce, err := redisNoncestore.SyncNetworkNonce(ctx, ko.MustString("system.public_key"))
-		lo.Debug("initNoncestore: syncing system nonce", "nonce", nonce)
-		if err != nil {
-			lo.Fatal("initNonceStore", "error", "system account nonce sync failed")
-		}
-	}
-
-	loadedPrivateKey, err := eth_crypto.HexToECDSA(ko.MustString("system.private_key"))
-	if err != nil {
-		lo.Fatal("Failed to load system private key", "error", err)
-	}
-
-	return loadedPrivateKey
-}

cmd/init_tasker.go

@@ -1,65 +0,0 @@
-package main
-
-import (
-	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
-	"github.com/grassrootseconomics/cic-custodial/internal/tasker/task"
-	"github.com/hibiken/asynq"
-)
-
-func initTasker() *tasker.TaskerServer {
-	lo.Debug("Bootstrapping tasker")
-
-	taskerServerOpts := tasker.TaskerServerOpts{
-		Concurrency:     ko.MustInt("asynq.concurrency"),
-		Logg:            lo,
-		RedisPool:       asynqRedisPool,
-		SystemContainer: nil,
-		TaskerClient:    taskerClient,
-	}
-
-	if debugFlag {
-		taskerServerOpts.LogLevel = asynq.DebugLevel
-	}
-
-	taskerServer := tasker.NewTaskerServer(taskerServerOpts)
-
-	taskerServer.RegisterHandlers(tasker.PrepareAccountTask, task.PrepareAccount(
-		redisNoncestore,
-		taskerClient,
-	))
-	taskerServer.RegisterHandlers(tasker.GiftGasTask, task.GiftGasProcessor(
-		celoProvider,
-		redisNoncestore,
-		lockProvider,
-		system,
-		taskerClient,
-	))
-	taskerServer.RegisterHandlers(tasker.GiftTokenTask, task.GiftTokenProcessor(
-		celoProvider,
-		redisNoncestore,
-		lockProvider,
-		system,
-		taskerClient,
-	))
-	taskerServer.RegisterHandlers(tasker.RefillGasTask, task.RefillGasProcessor(
-		celoProvider,
-		redisNoncestore,
-		lockProvider,
-		system,
-		taskerClient,
-	))
-	taskerServer.RegisterHandlers(tasker.TransferTokenTask, task.TransferToken(
-		celoProvider,
-		redisNoncestore,
-		postgresKeystore,
-		lockProvider,
-		system,
-		taskerClient,
-	))
-	taskerServer.RegisterHandlers(tasker.TxDispatchTask, task.TxDispatch(
-		celoProvider,
-	))
-
-	lo.Debug("Registered all tasker handlers")
-	return taskerServer
-}

cmd/main.go

@@ -1,122 +0,0 @@
-package main
-
-import (
-	"context"
-	"flag"
-	"os"
-	"os/signal"
-	"strings"
-	"sync"
-	"syscall"
-
-	"github.com/bsm/redislock"
-	celo "github.com/grassrootseconomics/cic-celo-sdk"
-	"github.com/grassrootseconomics/cic-custodial/internal/keystore"
-	"github.com/grassrootseconomics/cic-custodial/internal/nonce"
-	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
-	"github.com/grassrootseconomics/cic-custodial/pkg/redis"
-	"github.com/jackc/pgx/v5/pgxpool"
-	"github.com/knadh/koanf"
-	"github.com/labstack/echo/v4"
-	"github.com/zerodha/logf"
-)
-
-var (
-	confFlag       string
-	debugFlag      bool
-	taskerModeFlag bool
-	apiModeFlag    bool
-
-	asynqRedisPool   *redis.RedisPool
-	celoProvider     *celo.Provider
-	commonRedisPool  *redis.RedisPool
-	lo               logf.Logger
-	lockProvider     *redislock.Client
-	ko               *koanf.Koanf
-	postgresPool     *pgxpool.Pool
-	postgresKeystore keystore.Keystore
-	redisNoncestore  nonce.Noncestore
-	system           *tasker.SystemContainer
-	taskerClient     *tasker.TaskerClient
-)
-
-func init() {
-	flag.StringVar(&confFlag, "config", "config.toml", "Config file location")
-	flag.BoolVar(&debugFlag, "log", false, "Enable debug logging")
-	flag.BoolVar(&taskerModeFlag, "tasker", true, "Start tasker")
-	flag.BoolVar(&apiModeFlag, "api", true, "Start API server")
-	flag.Parse()
-
-	lo = initLogger(debugFlag)
-	ko = initConfig(confFlag)
-
-	celoProvider = initCeloProvider()
-	postgresPool = initPostgresPool()
-	postgresKeystore = initKeystore()
-	asynqRedisPool = initAsynqRedisPool()
-	commonRedisPool = initCommonRedisPool()
-	redisNoncestore = initRedisNoncestore()
-	lockProvider = initLockProvider()
-	taskerClient = initTaskerClient()
-	system = initSystemContainer()
-}
-
-func main() {
-	var (
-		tasker    *tasker.TaskerServer
-		apiServer *echo.Echo
-		wg        sync.WaitGroup
-	)
-
-	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
-	defer stop()
-
-	if apiModeFlag {
-		apiServer = initApiServer()
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-
-			lo.Info("Starting API server")
-			if err := apiServer.Start(ko.MustString("service.address")); err != nil {
-				if strings.Contains(err.Error(), "Server closed") {
-					lo.Info("Shutting down server")
-				} else {
-					lo.Fatal("Could not start api server", "err", err)
-				}
-			}
-		}()
-	}
-
-	if taskerModeFlag {
-		tasker = initTasker()
-
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-
-			lo.Info("Starting tasker")
-			if err := tasker.Start(); err != nil {
-				lo.Fatal("Could not start task server", "err", err)
-			}
-		}()
-	}
-
-	<-ctx.Done()
-	lo.Info("Graceful shutdown triggered")
-
-	if taskerModeFlag {
-		lo.Debug("Stopping tasker")
-		tasker.Stop()
-	}
-
-	if apiModeFlag {
-		lo.Debug("Stopping api server")
-		if err := apiServer.Shutdown(ctx); err != nil {
-			lo.Error("Could not gracefully shutdown api server", "err", err)
-		}
-	}
-
-	wg.Wait()
-}

cmd/service/api.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"github.com/VictoriaMetrics/metrics"
+	"github.com/go-playground/validator"
+	"github.com/grassrootseconomics/cic-custodial/internal/api"
+	"github.com/labstack/echo/v4"
+)
+
+// Bootstrap API server.
+func initApiServer(custodialContainer *custodial) *echo.Echo {
+	lo.Debug("api: bootstrapping api server")
+	server := echo.New()
+	server.HideBanner = true
+	server.HidePort = true
+
+	if ko.Bool("service.metrics") {
+		server.GET("/metrics", func(c echo.Context) error {
+			metrics.WritePrometheus(c.Response(), true)
+			return nil
+		})
+	}
+
+	server.Validator = &api.Validator{
+		ValidatorProvider: validator.New(),
+	}
+
+	apiRoute := server.Group("/api")
+	apiRoute.POST("/account/create", api.CreateAccountHandler(
+		custodialContainer.taskerClient,
+		custodialContainer.keystore,
+	))
+
+	return server
+}

cmd/service/custodial.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	"context"
+	"math/big"
+	"time"
+
+	eth_crypto "github.com/celo-org/celo-blockchain/crypto"
+	"github.com/go-redis/redis/v8"
+	"github.com/grassrootseconomics/cic-custodial/internal/nonce"
+	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
+	"github.com/grassrootseconomics/w3-celo-patch"
+)
+
+// Define common smart contrcat ABI's that can be injected into the system container.
+// Any relevant function signature that will be used by the custodial system can be defined here.
+func initAbis() map[string]*w3.Func {
+	return map[string]*w3.Func{
+		// Keccak hash -> 0x449a52f8
+		"mintTo": w3.MustNewFunc("mintTo(address, uint256)", "bool"),
+		// Keccak hash -> 0xa9059cbb
+		"transfer":     w3.MustNewFunc("transfer(address,uint256)", "bool"),
+		// Keccak hash -> 0x23b872dd
+		"transferFrom": w3.MustNewFunc("transferFrom(address, address, uint256)", "bool"),
+	}
+}
+
+// Bootstrap the internal custodial system configs and system signer key.
+// This container is passed down to individual tasker and API handlers.
+func initSystemContainer(ctx context.Context, noncestore nonce.Noncestore) (*tasker.SystemContainer, error) {
+	// Some custodial system defaults loaded from the config file.
+	systemContainer := &tasker.SystemContainer{
+		Abis:                  initAbis(),
+		GasRefillThreshold:    big.NewInt(ko.MustInt64("system.gas_refill_threshold")),
+		GasRefillValue:        big.NewInt(ko.MustInt64("system.gas_refill_value")),
+		GiftableGasValue:      big.NewInt(ko.MustInt64("system.giftable_gas_value")),
+		GiftableToken:         w3.A(ko.MustString("system.giftable_token_address")),
+		GiftableTokenValue:    big.NewInt(ko.MustInt64("system.giftable_token_value")),
+		LockPrefix:            ko.MustString("system.lock_prefix"),
+		LockTimeout:           1 * time.Second,
+		PublicKey:             ko.MustString("system.public_key"),
+		TokenDecimals:         ko.MustInt("system.token_decimals"),
+		TokenTransferGasLimit: uint64(ko.MustInt64("system.token_transfer_gas_limit")),
+	}
+	// Check if system signer account nonce is present.
+	// If not, we bootstrap it from the network.
+	currentSystemNonce, err := noncestore.Peek(ctx, ko.MustString("system.public_key"))
+	lo.Info("custodial: loaded (noncestore) system nonce", "nonce", currentSystemNonce)
+	if err == redis.Nil {
+		nonce, err := noncestore.SyncNetworkNonce(ctx, ko.MustString("system.public_key"))
+		lo.Info("custodial: syncing system nonce", "nonce", nonce)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	loadedPrivateKey, err := eth_crypto.HexToECDSA(ko.MustString("system.private_key"))
+	if err != nil {
+		return nil, err
+	}
+	systemContainer.PrivateKey = loadedPrivateKey
+
+	return systemContainer, nil
+}

cmd/service/init.go

@@ -20,6 +20,7 @@ import (
 	"github.com/zerodha/logf"
 )
 
+// Load config file.
 func initConfig(configFilePath string) *koanf.Koanf {
 	var (
 		ko = koanf.New(".")
@@ -40,6 +41,7 @@ func initConfig(configFilePath string) *koanf.Koanf {
 	return ko
 }
 
+// Load logger.
 func initLogger(debug bool) logf.Logger {
 	loggOpts := logg.LoggOpts{
 		Color: true,
@@ -53,7 +55,8 @@ func initLogger(debug bool) logf.Logger {
 	return logg.NewLogg(loggOpts)
 }
 
-func initCeloProvider() *celo.Provider {
+// Load Celo chain provider.
+func initCeloProvider() (*celo.Provider, error) {
 	providerOpts := celo.ProviderOpts{
 		RpcEndpoint: ko.MustString("chain.rpc_endpoint"),
 	}
@@ -66,38 +69,28 @@ func initCeloProvider() *celo.Provider {
 
 	provider, err := celo.NewProvider(providerOpts)
 	if err != nil {
-		lo.Fatal("initChainProvider", "error", err)
+		return nil, err
 	}
 
-	return provider
+	return provider, nil
 }
 
-func initPostgresPool() *pgxpool.Pool {
+// Load postgres pool.
+func initPostgresPool() (*pgxpool.Pool, error) {
 	poolOpts := postgres.PostgresPoolOpts{
 		DSN: ko.MustString("postgres.dsn"),
 	}
 
 	pool, err := postgres.NewPostgresPool(poolOpts)
 	if err != nil {
-		lo.Fatal("initPostgresPool", "error", err)
+		return nil, err
 	}
 
-	return pool
+	return pool, nil
 }
 
-func initKeystore() keystore.Keystore {
+// Load separate redis connection for the tasker on a reserved db namespace.
-	keystore, err := keystore.NewPostgresKeytore(keystore.Opts{
+func initAsynqRedisPool() (*redis.RedisPool, error) {
-		PostgresPool: postgresPool,
-		Logg:         lo,
-	})
-	if err != nil {
-		lo.Fatal("initKeystore", "error", err)
-	}
-
-	return keystore
-}
-
-func initAsynqRedisPool() *redis.RedisPool {
 	poolOpts := redis.RedisPoolOpts{
 		DSN:          ko.MustString("asynq.dsn"),
 		MinIdleConns: ko.MustInt("redis.minconn"),
@@ -105,13 +98,14 @@ func initAsynqRedisPool() *redis.RedisPool {
 
 	pool, err := redis.NewRedisPool(poolOpts)
 	if err != nil {
-		lo.Fatal("initAsynqRedisPool", "error", err)
+		return nil, err
 	}
 
-	return pool
+	return pool, nil
 }
 
-func initCommonRedisPool() *redis.RedisPool {
+// Common redis connection on a different db namespace from the takser.
+func initCommonRedisPool() (*redis.RedisPool, error) {
 	poolOpts := redis.RedisPoolOpts{
 		DSN:          ko.MustString("redis.dsn"),
 		MinIdleConns: ko.MustInt("redis.minconn"),
@@ -119,26 +113,42 @@ func initCommonRedisPool() *redis.RedisPool {
 
 	pool, err := redis.NewRedisPool(poolOpts)
 	if err != nil {
-		lo.Fatal("initCommonRedisPool", "error", err)
+		return nil, err
 	}
 
-	return pool
+	return pool, nil
 }
 
-func initRedisNoncestore() nonce.Noncestore {
+// Load postgres based keystore
+func initPostgresKeystore(postgresPool *pgxpool.Pool) (keystore.Keystore, error) {
+	keystore, err := keystore.NewPostgresKeytore(keystore.Opts{
+		PostgresPool: postgresPool,
+		Logg:         lo,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return keystore, nil
+}
+
+// Load redis backed noncestore.
+func initRedisNoncestore(redisPool *redis.RedisPool, celoProvider *celo.Provider) nonce.Noncestore {
 	return nonce.NewRedisNoncestore(nonce.Opts{
-		RedisPool:     commonRedisPool,
+		RedisPool:    redisPool,
-		ChainProvider: celoProvider,
+		CeloProvider: celoProvider,
 	})
 }
 
-func initLockProvider() *redislock.Client {
+// Load global lock provider.
-	return redislock.New(commonRedisPool.Client)
+func initLockProvider(redisPool redislock.RedisClient) *redislock.Client {
+	return redislock.New(redisPool)
 }
 
-func initTaskerClient() *tasker.TaskerClient {
+// Load tasker client.
+func initTaskerClient(redisPool *redis.RedisPool) *tasker.TaskerClient {
 	return tasker.NewTaskerClient(tasker.TaskerClientOpts{
-		RedisPool:     asynqRedisPool,
+		RedisPool:     redisPool,
-		TaskRetention: time.Hour * 12,
+		TaskRetention: time.Duration(ko.MustInt64("asynq.task_retention_hrs")) * time.Hour,
 	})
 }

cmd/service/main.go

@@ -0,0 +1,136 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"syscall"
+
+	"github.com/bsm/redislock"
+	celo "github.com/grassrootseconomics/cic-celo-sdk"
+	"github.com/grassrootseconomics/cic-custodial/internal/keystore"
+	"github.com/grassrootseconomics/cic-custodial/internal/nonce"
+	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
+	"github.com/knadh/koanf"
+	"github.com/labstack/echo/v4"
+	"github.com/zerodha/logf"
+)
+
+var (
+	confFlag  string
+	debugFlag bool
+
+	lo logf.Logger
+	ko *koanf.Koanf
+)
+
+type custodial struct {
+	celoProvider    *celo.Provider
+	keystore        keystore.Keystore
+	lockProvider    *redislock.Client
+	noncestore      nonce.Noncestore
+	systemContainer *tasker.SystemContainer
+	taskerClient    *tasker.TaskerClient
+}
+
+func init() {
+	flag.StringVar(&confFlag, "config", "config.toml", "Config file location")
+	flag.BoolVar(&debugFlag, "log", false, "Enable debug logging")
+	flag.Parse()
+
+	lo = initLogger(debugFlag)
+	ko = initConfig(confFlag)
+}
+
+func main() {
+	var (
+		tasker    *tasker.TaskerServer
+		apiServer *echo.Echo
+		wg        sync.WaitGroup
+	)
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	celoProvider, err := initCeloProvider()
+	if err != nil {
+		lo.Fatal("main: critical error loading chain provider", "error", err)
+	}
+
+	postgresPool, err := initPostgresPool()
+	if err != nil {
+		lo.Fatal("main: critical error connecting to postgres", "error", err)
+	}
+
+	asynqRedisPool, err := initAsynqRedisPool()
+	if err != nil {
+		lo.Fatal("main: critical error connecting to asynq redis db", "error", err)
+	}
+
+	redisPool, err := initCommonRedisPool()
+	if err != nil {
+		lo.Fatal("main: critical error connecting to common redis db", "error", err)
+	}
+
+	postgresKeystore, err := initPostgresKeystore(postgresPool)
+	if err != nil {
+		lo.Fatal("main: critical error loading postgres keystore", "error", err)
+	}
+
+	redisNoncestore := initRedisNoncestore(redisPool, celoProvider)
+	lockProvider := initLockProvider(redisPool.Client)
+	taskerClient := initTaskerClient(asynqRedisPool)
+
+	systemContainer, err := initSystemContainer(context.Background(), redisNoncestore)
+	if err != nil {
+		lo.Fatal("main: critical error bootstrapping system container", "error", err)
+	}
+
+	custodial := &custodial{
+		celoProvider:    celoProvider,
+		keystore:        postgresKeystore,
+		lockProvider:    lockProvider,
+		noncestore:      redisNoncestore,
+		systemContainer: systemContainer,
+		taskerClient:    taskerClient,
+	}
+
+	apiServer = initApiServer(custodial)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		lo.Info("main: starting API server")
+		if err := apiServer.Start(ko.MustString("service.address")); err != nil {
+			if strings.Contains(err.Error(), "Server closed") {
+				lo.Info("main: shutting down server")
+			} else {
+				lo.Fatal("main: critical error shutting down server", "err", err)
+			}
+		}
+	}()
+
+	tasker = initTasker(custodial, asynqRedisPool)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		lo.Info("Starting tasker")
+		if err := tasker.Start(); err != nil {
+			lo.Fatal("main: could not start task server", "err", err)
+		}
+	}()
+
+	<-ctx.Done()
+
+	lo.Info("main: stopping tasker")
+	tasker.Stop()
+
+	lo.Info("main: stopping api server")
+	if err := apiServer.Shutdown(ctx); err != nil {
+		lo.Error("Could not gracefully shutdown api server", "err", err)
+	}
+
+	wg.Wait()
+}

cmd/service/tasker.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
+	"github.com/grassrootseconomics/cic-custodial/internal/tasker/task"
+	"github.com/grassrootseconomics/cic-custodial/pkg/redis"
+	"github.com/hibiken/asynq"
+)
+
+// Load tasker handlers injecting necessary handler dependencies from the system container.
+func initTasker(custodialContainer *custodial, redisPool *redis.RedisPool) *tasker.TaskerServer {
+	lo.Debug("Bootstrapping tasker")
+
+	taskerServerOpts := tasker.TaskerServerOpts{
+		Concurrency:     ko.MustInt("asynq.worker_count"),
+		Logg:            lo,
+		RedisPool:       redisPool,
+		SystemContainer: custodialContainer.systemContainer,
+		TaskerClient:    custodialContainer.taskerClient,
+	}
+
+	if debugFlag {
+		taskerServerOpts.LogLevel = asynq.DebugLevel
+	}
+
+	taskerServer := tasker.NewTaskerServer(taskerServerOpts)
+
+	taskerServer.RegisterHandlers(tasker.PrepareAccountTask, task.PrepareAccount(
+		custodialContainer.noncestore,
+		custodialContainer.taskerClient,
+	))
+	taskerServer.RegisterHandlers(tasker.GiftGasTask, task.GiftGasProcessor(
+		custodialContainer.celoProvider,
+		custodialContainer.noncestore,
+		custodialContainer.lockProvider,
+		custodialContainer.systemContainer,
+		custodialContainer.taskerClient,
+	))
+	taskerServer.RegisterHandlers(tasker.GiftTokenTask, task.GiftTokenProcessor(
+		custodialContainer.celoProvider,
+		custodialContainer.noncestore,
+		custodialContainer.lockProvider,
+		custodialContainer.systemContainer,
+		custodialContainer.taskerClient,
+	))
+	taskerServer.RegisterHandlers(tasker.RefillGasTask, task.RefillGasProcessor(
+		custodialContainer.celoProvider,
+		custodialContainer.noncestore,
+		custodialContainer.lockProvider,
+		custodialContainer.systemContainer,
+		custodialContainer.taskerClient,
+	))
+	taskerServer.RegisterHandlers(tasker.TransferTokenTask, task.TransferToken(
+		custodialContainer.celoProvider,
+		custodialContainer.noncestore,
+		custodialContainer.keystore,
+		custodialContainer.lockProvider,
+		custodialContainer.systemContainer,
+		custodialContainer.taskerClient,
+	))
+	taskerServer.RegisterHandlers(tasker.TxDispatchTask, task.TxDispatch(
+		custodialContainer.celoProvider,
+	))
+
+	return taskerServer
+}

config.toml

@@ -1,33 +1,45 @@
 [service]
-address          = ":5000"
+address = ":5000"
-statsviz_debug   = true
+# Exposes Prometheus metrics
+# /metrics endpoint
+metrics = true
 
+# System default values
+# Valus are in wei unless otherwise stated
 [system]
-gas_refill_threshold     = 100000000000000
+# Any account below 1 KES equivalent of CELO is topped up again
-gas_refill_value         = 100000000000000
+# 10000000000000000 = 1 KES
-giftable_gas_value       = 200000000000000
+gas_refill_threshold = 10000000000000000
-giftable_token_address   = "0x486aD10d70107900546455F7a0e022c300F157Bf"
+gas_refill_value = 10000000000000000
-giftable_token_value     = 5000000000000000000
+# Every custodial account is given 2 KES worth of CELO
-private_key              = "a6af6c597c614e3c8ee4b7638ab7c3f737aece3773a5413ca8caf4338e6b06d1"
+giftable_gas_value = 20000000000000000
-lock_prefix              = "lock:"
+# The giftable token is a training voucher
-public_key               = "0x80097c773B3E83472FC7952c5206a7DB35d42bEF"
+# Every new user is given 5 DGFT
-token_decimals           = 18
+giftable_token_address = "0x486aD10d70107900546455F7a0e022c300F157Bf"
+giftable_token_value = 5000000
+# System private key
+# Should always be toped up
+private_key = "a6af6c597c614e3c8ee4b7638ab7c3f737aece3773a5413ca8caf4338e6b06d1"
+lock_prefix = "lock:"
+public_key = "0x80097c773B3E83472FC7952c5206a7DB35d42bEF"
+token_decimals = 6
 token_transfer_gas_limit = 100000
 
 [chain]
 rpc_endpoint = "https://alfajores-forno.celo-testnet.org"
-testnet      = true
+testnet = true
 
 [postgres]
 debug = false
-dsn   = "postgres://postgres:postgres@localhost:5432/cic_custodial"
+dsn = "postgres://postgres:postgres@localhost:5432/cic_custodial"
 
 [redis]
-debug   = false
+debug = false
-dsn     = "redis://localhost:6379/1"
+dsn = "redis://localhost:6379/1"
-minconn = 5
+min_idle_conn = 5
 
 [asynq]
-concurrency = 25
+worker_count = 15
-debug       = false
+debug = false
-dsn         = "redis://redis:6379/0"
+dsn = "redis://redis:6379/0"
+task_retention_hrs = 24

go.mod

@@ -22,6 +22,7 @@ require (
 require (
 	filippo.io/edwards25519 v1.0.0 // indirect
 	github.com/VictoriaMetrics/fastcache v1.12.0 // indirect
+	github.com/VictoriaMetrics/metrics v1.23.1 // indirect
 	github.com/btcsuite/btcd v0.20.1-beta // indirect
 	github.com/celo-org/celo-bls-go v0.6.4 // indirect
 	github.com/celo-org/celo-bls-go-android v0.6.3 // indirect
@@ -74,7 +75,9 @@ require (
 	github.com/tklauser/go-sysconf v0.3.11 // indirect
 	github.com/tklauser/numcpus v0.6.0 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
+	github.com/valyala/fastrand v1.1.0 // indirect
 	github.com/valyala/fasttemplate v1.2.2 // indirect
+	github.com/valyala/histogram v1.2.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	golang.org/x/crypto v0.3.0 // indirect

go.sum

@@ -43,6 +43,8 @@ github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrU
 github.com/VictoriaMetrics/fastcache v1.6.0/go.mod h1:0qHz5QP0GMX4pfmMA/zt5RgfNuXJrTP0zS7DqpHGGTw=
 github.com/VictoriaMetrics/fastcache v1.12.0 h1:vnVi/y9yKDcD9akmc4NqAoqgQhJrOwUF+j9LTgn4QDE=
 github.com/VictoriaMetrics/fastcache v1.12.0/go.mod h1:tjiYeEfYXCqacuvYw/7UoDIeJaNxq6132xHICNP77w8=
+github.com/VictoriaMetrics/metrics v1.23.1 h1:/j8DzeJBxSpL2qSIdqnRFLvQQhbJyJbbEi22yMm7oL0=
+github.com/VictoriaMetrics/metrics v1.23.1/go.mod h1:rAr/llLpEnAdTehiNlUxKgnjcOuROSzpw0GvjpEbvFc=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
 github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -605,10 +607,14 @@ github.com/tyler-smith/go-bip39 v1.0.1-0.20181017060643-dbb3b84ba2ef/go.mod h1:s
 github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
+github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8=
+github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
 github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
 github.com/valyala/fasttemplate v1.2.1/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo=
 github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
+github.com/valyala/histogram v1.2.0 h1:wyYGAZZt3CpwUiIb9AU/Zbllg1llXyrtApRS815OLoQ=
+github.com/valyala/histogram v1.2.0/go.mod h1:Hb4kBwb4UxsaNbbbh+RRz8ZR6pdodR57tzWUS3BUzXY=
 github.com/willf/bitset v1.1.3/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

internal/api/account.go

@@ -0,0 +1,47 @@
+package api
+
+import (
+	"net/http"
+
+	"github.com/grassrootseconomics/cic-custodial/internal/keystore"
+	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
+	"github.com/grassrootseconomics/cic-custodial/pkg/keypair"
+	"github.com/labstack/echo/v4"
+)
+
+func CreateAccountHandler(
+	taskerClient *tasker.TaskerClient,
+	keystore keystore.Keystore,
+) func(echo.Context) error {
+	return func(c echo.Context) error {
+		generatedKeyPair, err := keypair.Generate()
+		if err != nil {
+			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
+				Ok:   false,
+				Code: INTERNAL_ERROR,
+			})
+		}
+
+		if err := keystore.WriteKeyPair(c.Request().Context(), generatedKeyPair); err != nil {
+			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
+				Ok:   false,
+				Code: INTERNAL_ERROR,
+			})
+		}
+
+		return c.JSON(http.StatusOK, okResp{
+			Ok: true,
+			Result: H{
+				"publicKey": generatedKeyPair.Public,
+			},
+		})
+	}
+}
+
+func AccountStatusHandler() func(echo.Context) error {
+	return func(c echo.Context) error {
+		return c.JSON(http.StatusOK, okResp{
+			Ok: true,
+		})
+	}
+}

internal/api/registration.go

@@ -1,71 +0,0 @@
-package api
-
-import (
-	"encoding/json"
-	"net/http"
-
-	"github.com/grassrootseconomics/cic-custodial/internal/keystore"
-	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
-	"github.com/grassrootseconomics/cic-custodial/internal/tasker/task"
-	"github.com/grassrootseconomics/cic-custodial/pkg/keypair"
-	"github.com/labstack/echo/v4"
-)
-
-type registrationResponse struct {
-	PublicKey string `json:"publicKey"`
-	TaskRef    string `json:"taskRef"`
-}
-
-func RegistrationHandler(
-	taskerClient *tasker.TaskerClient,
-	keystore keystore.Keystore,
-) func(echo.Context) error {
-	return func(c echo.Context) error {
-		generatedKeyPair, err := keypair.Generate()
-		if err != nil {
-			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
-				Ok:    false,
-				Error: KEYPAIR_ERROR,
-			})
-		}
-
-		if err := keystore.WriteKeyPair(c.Request().Context(), generatedKeyPair); err != nil {
-			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
-				Ok:    false,
-				Error: INTERNAL_ERROR,
-			})
-		}
-
-		taskPayload, err := json.Marshal(task.SystemPayload{
-			PublicKey: generatedKeyPair.Public,
-		})
-		if err != nil {
-			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
-				Ok:    false,
-				Error: JSON_MARSHAL_ERROR,
-			})
-		}
-
-		task, err := taskerClient.CreateTask(
-			tasker.PrepareAccountTask,
-			tasker.DefaultPriority,
-			&tasker.Task{
-				Payload: taskPayload,
-			},
-		)
-		if err != nil {
-			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
-				Ok:    false,
-				Error: TASK_CHAIN_ERROR,
-			})
-		}
-
-		return c.JSON(http.StatusOK, okResp{
-			Ok: true,
-			Data: registrationResponse{
-				PublicKey: generatedKeyPair.Public,
-				TaskRef:    task.ID,
-			},
-		})
-	}
-}

internal/api/transfer.go

@@ -1,69 +0,0 @@
-package api
-
-import (
-	"encoding/json"
-	"net/http"
-
-	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
-	"github.com/labstack/echo/v4"
-)
-
-type (
-	transferRequest struct {
-		From           string `json:"from" validate:"required,eth_addr"`
-		To             string `json:"to" validate:"required,eth_addr"`
-		VoucherAddress string `json:"voucherAddress" validate:"required,eth_addr"`
-		Amount         string `json:"amount" validate:"required,numeric"`
-	}
-
-	transferResponse struct {
-		TaskRef string `json:"taskRef"`
-	}
-)
-
-func TransferHandler(
-	taskerClient *tasker.TaskerClient,
-) func(echo.Context) error {
-	return func(c echo.Context) error {
-		transferPayload := new(transferRequest)
-
-		if err := c.Bind(transferPayload); err != nil {
-			return echo.NewHTTPError(http.StatusBadRequest, errResp{
-				Ok:    false,
-				Error: BIND_ERROR,
-			})
-		}
-		if err := c.Validate(transferPayload); err != nil {
-			return err
-		}
-
-		taskPayload, err := json.Marshal(transferPayload)
-		if err != nil {
-			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
-				Ok:    false,
-				Error: JSON_MARSHAL_ERROR,
-			})
-		}
-
-		task, err := taskerClient.CreateTask(
-			tasker.TransferTokenTask,
-			tasker.DefaultPriority,
-			&tasker.Task{
-				Payload: taskPayload,
-			},
-		)
-		if err != nil {
-			return echo.NewHTTPError(http.StatusInternalServerError, errResp{
-				Ok:    false,
-				Error: TASK_CHAIN_ERROR,
-			})
-		}
-
-		return c.JSON(http.StatusOK, okResp{
-			Ok: true,
-			Data: transferResponse{
-				TaskRef: task.ID,
-			},
-		})
-	}
-}

internal/api/types.go

@@ -1,20 +1,18 @@
 package api
 
 const (
-	INTERNAL_ERROR     = "ERR_INTERNAL"
+	INTERNAL_ERROR   = "ERR_INTERNAL"
-	KEYPAIR_ERROR      = "ERR_GEN_KEYPAIR"
+	VALIDATION_ERROR = "ERR_VALIDATE"
-	JSON_MARSHAL_ERROR = "ERR_PAYLOAD_SERIALIZATION"
-	TASK_CHAIN_ERROR   = "ERR_START_TASK_CHAIN"
-	VALIDATION_ERROR   = "ERR_VALIDATE"
-	BIND_ERROR         = "ERR_BIND"
 )
 
+type H map[string]any
+
 type okResp struct {
-	Ok   bool        `json:"ok"`
+	Ok     bool `json:"ok"`
-	Data interface{} `json:"data"`
+	Result H    `json:"result"`
 }
 
 type errResp struct {
-	Ok    bool   `json:"ok"`
+	Ok   bool   `json:"ok"`
-	Error string `json:"error"`
+	Code string `json:"errorCode"`
 }

internal/api/validator.go

@@ -1,23 +1,21 @@
 package api
 
 import (
-	"fmt"
 	"net/http"
 
 	"github.com/go-playground/validator"
 	"github.com/labstack/echo/v4"
 )
 
-type CustomValidator struct {
+type Validator struct {
-	Validator *validator.Validate
+	ValidatorProvider *validator.Validate
 }
 
-func (cv *CustomValidator) Validate(i interface{}) error {
+func (v *Validator) Validate(i interface{}) error {
-	if err := cv.Validator.Struct(i); err != nil {
+	if err := v.ValidatorProvider.Struct(i); err != nil {
-		fmt.Println(err)
 		return echo.NewHTTPError(http.StatusBadRequest, errResp{
-			Ok:    false,
+			Ok:   false,
-			Error: VALIDATION_ERROR,
+			Code: VALIDATION_ERROR,
 		})
 	}
 	return nil

internal/nonce/redis.go

@@ -10,8 +10,8 @@ import (
 )
 
 type Opts struct {
-	RedisPool     *redispool.RedisPool
+	RedisPool    *redispool.RedisPool
-	ChainProvider *celo.Provider
+	CeloProvider *celo.Provider
 }
 
 // RedisNoncestore implements `Noncestore`
@@ -23,7 +23,7 @@ type RedisNoncestore struct {
 func NewRedisNoncestore(o Opts) Noncestore {
 	return &RedisNoncestore{
 		redis:         o.RedisPool,
-		chainProvider: o.ChainProvider,
+		chainProvider: o.CeloProvider,
 	}
 }
 

pkg/keypair/keypair.go

@@ -12,6 +12,7 @@ type Key struct {
 	Private string
 }
 
+// Generate creates a new keypair from internally randomized entropy.
 func Generate() (Key, error) {
 	privateKey, err := crypto.GenerateKey()
 	if err != nil {