major refactor: use proxy contract and gas faucet (see notes)

* remove uncessary tasks and task handlers
* reafctor custodial container
* refactor gas refiller. Gas refiller can queue at a later time to match cooldown
* refactor sub handler to process chain events
This commit is contained in:
Mohamed Sohail 2023-03-29 16:10:58 +00:00
parent 448b142f7c
commit e203c49049
Signed by: kamikazechaser
GPG Key ID: 7DD45520C01CD85D
27 changed files with 339 additions and 730 deletions

View File

@ -1,73 +0,0 @@
package main
import (
"context"
"math/big"
"time"
eth_crypto "github.com/celo-org/celo-blockchain/crypto"
"github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/nonce"
"github.com/grassrootseconomics/w3-celo-patch"
"github.com/redis/go-redis/v9"
)
// 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"),
// Add to account index
"add": w3.MustNewFunc("add(address)", "bool"),
// giveTo gas refill
"giveTo": w3.MustNewFunc("giveTo(address)", "uint256"),
}
}
// 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) *custodial.SystemContainer {
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
// Some custodial system defaults loaded from the config file.
systemContainer := &custodial.SystemContainer{
Abis: initAbis(),
AccountIndexContract: w3.A(ko.MustString("system.account_index_address")),
GasFaucetContract: w3.A(ko.MustString("system.gas_faucet_address")),
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")),
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 (first boot), we bootstrap it from the network.
currentSystemNonce, err := noncestore.Peek(ctx, ko.MustString("system.public_key"))
lo.Info("custodial: loaded system nonce from noncestore", "nonce", currentSystemNonce)
if err == redis.Nil {
nonce, err := noncestore.SyncNetworkNonce(ctx, ko.MustString("system.public_key"))
lo.Info("custodial: syncing system nonce from network", "nonce", nonce)
if err != nil {
lo.Fatal("custodial: critical error bootstrapping system container", "error", err)
}
}
loadedPrivateKey, err := eth_crypto.HexToECDSA(ko.MustString("system.private_key"))
if err != nil {
lo.Fatal("custodial: critical error bootstrapping system container", "error", err)
}
systemContainer.PrivateKey = loadedPrivateKey
return systemContainer
}

View File

@ -157,10 +157,9 @@ func initPostgresKeystore(postgresPool *pgxpool.Pool, queries *queries.Queries)
} }
// Load redis backed noncestore. // Load redis backed noncestore.
func initRedisNoncestore(redisPool *redis.RedisPool, celoProvider *celoutils.Provider) nonce.Noncestore { func initRedisNoncestore(redisPool *redis.RedisPool) nonce.Noncestore {
return nonce.NewRedisNoncestore(nonce.Opts{ return nonce.NewRedisNoncestore(nonce.Opts{
RedisPool: redisPool, RedisPool: redisPool,
CeloProvider: celoProvider,
}) })
} }

View File

@ -14,13 +14,11 @@ import (
"github.com/zerodha/logf" "github.com/zerodha/logf"
) )
type ( type internalServicesContainer struct {
internalServiceContainer struct {
apiService *echo.Echo apiService *echo.Echo
jetstreamSub *sub.Sub jetstreamSub *sub.Sub
taskerService *tasker.TaskerServer taskerService *tasker.TaskerServer
} }
)
var ( var (
build string build string
@ -56,15 +54,14 @@ func main() {
postgresKeystore := initPostgresKeystore(postgresPool, parsedQueries) postgresKeystore := initPostgresKeystore(postgresPool, parsedQueries)
pgStore := initPostgresStore(postgresPool, parsedQueries) pgStore := initPostgresStore(postgresPool, parsedQueries)
redisNoncestore := initRedisNoncestore(redisPool, celoProvider) redisNoncestore := initRedisNoncestore(redisPool)
lockProvider := initLockProvider(redisPool.Client) lockProvider := initLockProvider(redisPool.Client)
taskerClient := initTaskerClient(asynqRedisPool) taskerClient := initTaskerClient(asynqRedisPool)
systemContainer := initSystemContainer(context.Background(), redisNoncestore)
natsConn, jsCtx := initJetStream() natsConn, jsCtx := initJetStream()
jsPub := initPub(jsCtx) jsPub := initPub(jsCtx)
custodial := &custodial.Custodial{ custodial, err := custodial.NewCustodial(custodial.Opts{
CeloProvider: celoProvider, CeloProvider: celoProvider,
Keystore: postgresKeystore, Keystore: postgresKeystore,
LockProvider: lockProvider, LockProvider: lockProvider,
@ -72,11 +69,16 @@ func main() {
PgStore: pgStore, PgStore: pgStore,
Pub: jsPub, Pub: jsPub,
RedisClient: redisPool.Client, RedisClient: redisPool.Client,
SystemContainer: systemContainer, RegistryAddress: ko.MustString("chain.registry_address"),
SystemPrivateKey: ko.MustString("system.private_key"),
SystemPublicKey: ko.MustString("system.public_key"),
TaskerClient: taskerClient, TaskerClient: taskerClient,
})
if err != nil {
lo.Fatal("main: crtical error loading custodial container", "error", err)
} }
internalServices := &internalServiceContainer{} internalServices := &internalServicesContainer{}
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
signalCh, closeCh := createSigChannel() signalCh, closeCh := createSigChannel()

View File

@ -34,11 +34,7 @@ func initTasker(custodialContainer *custodial.Custodial, redisPool *redis.RedisP
observibilityMiddleware(), observibilityMiddleware(),
}) })
taskerServer.RegisterHandlers(tasker.AccountPrepareTask, task.AccountPrepare(custodialContainer))
taskerServer.RegisterHandlers(tasker.AccountRegisterTask, task.AccountRegisterOnChainProcessor(custodialContainer)) taskerServer.RegisterHandlers(tasker.AccountRegisterTask, task.AccountRegisterOnChainProcessor(custodialContainer))
taskerServer.RegisterHandlers(tasker.AccountGiftGasTask, task.AccountGiftGasProcessor(custodialContainer))
taskerServer.RegisterHandlers(tasker.AccountGiftVoucherTask, task.GiftVoucherProcessor(custodialContainer))
taskerServer.RegisterHandlers(tasker.AccountActivateTask, task.AccountActivateProcessor(custodialContainer))
taskerServer.RegisterHandlers(tasker.AccountRefillGasTask, task.AccountRefillGasProcessor(custodialContainer)) taskerServer.RegisterHandlers(tasker.AccountRefillGasTask, task.AccountRefillGasProcessor(custodialContainer))
taskerServer.RegisterHandlers(tasker.SignTransferTask, task.SignTransfer(custodialContainer)) taskerServer.RegisterHandlers(tasker.SignTransferTask, task.SignTransfer(custodialContainer))
taskerServer.RegisterHandlers(tasker.DispatchTxTask, task.DispatchTx(custodialContainer)) taskerServer.RegisterHandlers(tasker.DispatchTxTask, task.DispatchTx(custodialContainer))

View File

@ -17,7 +17,7 @@ func createSigChannel() (chan os.Signal, func()) {
} }
} }
func startGracefulShutdown(ctx context.Context, internalServices *internalServiceContainer) { func startGracefulShutdown(ctx context.Context, internalServices *internalServicesContainer) {
ctx, cancel := context.WithTimeout(ctx, 5*time.Second) ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel() defer cancel()

View File

@ -8,27 +8,11 @@ metrics = true
rpc_endpoint = "" rpc_endpoint = ""
testnet = true testnet = true
devnet = false devnet = false
registry_address = ""
[system] [system]
# System default values
# Gas values are in wei with 18 d.p. precision unless otherwise stated
# Token values are in wei with 6 d.p. precision unless otherwise stated
# All addresses MUST be checksumed
account_index_address = ""
gas_faucet_address = ""
gas_refill_threshold = 2500000000000000
gas_refill_value = 15000000000000000
giftable_gas_value = 15000000000000000
giftable_token_address = ""
giftable_token_value = 5000000
# System private key
# Should always be toped up
private_key = "" private_key = ""
public_key = "" public_key = ""
token_decimals = 6
token_transfer_gas_limit = 200000
[postgres] [postgres]
dsn = "" dsn = ""

2
go.mod
View File

@ -9,7 +9,7 @@ require (
github.com/georgysavva/scany/v2 v2.0.0 github.com/georgysavva/scany/v2 v2.0.0
github.com/go-playground/validator/v10 v10.12.0 github.com/go-playground/validator/v10 v10.12.0
github.com/google/uuid v1.3.0 github.com/google/uuid v1.3.0
github.com/grassrootseconomics/celoutils v1.1.1 github.com/grassrootseconomics/celoutils v1.2.1
github.com/grassrootseconomics/w3-celo-patch v0.2.0 github.com/grassrootseconomics/w3-celo-patch v0.2.0
github.com/hibiken/asynq v0.24.0 github.com/hibiken/asynq v0.24.0
github.com/jackc/pgx/v5 v5.3.1 github.com/jackc/pgx/v5 v5.3.1

2
go.sum
View File

@ -262,6 +262,8 @@ github.com/grassrootseconomics/asynq v0.25.0 h1:2zSz5YwNLu/oCTm/xfNixn86i9aw4zth
github.com/grassrootseconomics/asynq v0.25.0/go.mod h1:pe2XOdK1eIbTgTmRFHIYl75lvVuTPJxZq2T9Ocz/+2s= github.com/grassrootseconomics/asynq v0.25.0/go.mod h1:pe2XOdK1eIbTgTmRFHIYl75lvVuTPJxZq2T9Ocz/+2s=
github.com/grassrootseconomics/celoutils v1.1.1 h1:REsndvfBkPN8UKOoQFNEGm/sCwKtTm+woYtgMl3bfZ0= github.com/grassrootseconomics/celoutils v1.1.1 h1:REsndvfBkPN8UKOoQFNEGm/sCwKtTm+woYtgMl3bfZ0=
github.com/grassrootseconomics/celoutils v1.1.1/go.mod h1:Uo5YRy6AGLAHDZj9jaOI+AWoQ1H3L0v79728pPMkm9Q= github.com/grassrootseconomics/celoutils v1.1.1/go.mod h1:Uo5YRy6AGLAHDZj9jaOI+AWoQ1H3L0v79728pPMkm9Q=
github.com/grassrootseconomics/celoutils v1.2.1 h1:ndM4h7Df0d57m2kdRXRStrnunqOL61wQ51rnOanX1KI=
github.com/grassrootseconomics/celoutils v1.2.1/go.mod h1:Uo5YRy6AGLAHDZj9jaOI+AWoQ1H3L0v79728pPMkm9Q=
github.com/grassrootseconomics/w3-celo-patch v0.2.0 h1:YqibbPzX0tQKmxU1nUGzThPKk/fiYeYZY6Aif3eyu8U= github.com/grassrootseconomics/w3-celo-patch v0.2.0 h1:YqibbPzX0tQKmxU1nUGzThPKk/fiYeYZY6Aif3eyu8U=
github.com/grassrootseconomics/w3-celo-patch v0.2.0/go.mod h1:WhBXNzNIvHmS6B2hAeShs56oa9Azb4jQSrOMKuMdBWw= github.com/grassrootseconomics/w3-celo-patch v0.2.0/go.mod h1:WhBXNzNIvHmS6B2hAeShs56oa9Azb4jQSrOMKuMdBWw=
github.com/hashicorp/go-bexpr v0.1.10/go.mod h1:oxlubA2vC/gFVfX1A6JGp7ls7uCDlfJn732ehYYg+g0= github.com/hashicorp/go-bexpr v0.1.10/go.mod h1:oxlubA2vC/gFVfX1A6JGp7ls7uCDlfJn732ehYYg+g0=

View File

@ -38,7 +38,7 @@ func HandleAccountCreate(cu *custodial.Custodial) func(echo.Context) error {
_, err = cu.TaskerClient.CreateTask( _, err = cu.TaskerClient.CreateTask(
c.Request().Context(), c.Request().Context(),
tasker.AccountPrepareTask, tasker.AccountRegisterTask,
tasker.DefaultPriority, tasker.DefaultPriority,
&tasker.Task{ &tasker.Task{
Id: trackingId, Id: trackingId,

View File

@ -5,6 +5,7 @@ import (
"math/big" "math/big"
"net/http" "net/http"
"github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/custodial" "github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/w3-celo-patch" "github.com/grassrootseconomics/w3-celo-patch"
"github.com/grassrootseconomics/w3-celo-patch/module/eth" "github.com/grassrootseconomics/w3-celo-patch/module/eth"
@ -31,8 +32,8 @@ func HandleNetworkAccountStatus(cu *custodial.Custodial) func(echo.Context) erro
if err := cu.CeloProvider.Client.CallCtx( if err := cu.CeloProvider.Client.CallCtx(
c.Request().Context(), c.Request().Context(),
eth.Nonce(w3.A(accountStatusRequest.Address), nil).Returns(&networkNonce), eth.Nonce(celoutils.HexToAddress(accountStatusRequest.Address), nil).Returns(&networkNonce),
eth.Balance(w3.A(accountStatusRequest.Address), nil).Returns(&networkBalance), eth.Balance(celoutils.HexToAddress(accountStatusRequest.Address), nil).Returns(&networkBalance),
); err != nil { ); err != nil {
return err return err
} }

View File

@ -52,14 +52,36 @@ func HandleSignTransfer(cu *custodial.Custodial) func(echo.Context) error {
}) })
} }
trackingId := uuid.NewString()
if gasQuota < 1 { if gasQuota < 1 {
gasRefillPayload, err := json.Marshal(task.AccountPayload{
PublicKey: req.From,
TrackingId: trackingId,
})
if err != nil {
return err
}
_, err = cu.TaskerClient.CreateTask(
c.Request().Context(),
tasker.AccountRefillGasTask,
tasker.DefaultPriority,
&tasker.Task{
Id: trackingId,
Payload: gasRefillPayload,
},
)
if err != nil {
return err
}
return c.JSON(http.StatusForbidden, ErrResp{ return c.JSON(http.StatusForbidden, ErrResp{
Ok: false, Ok: false,
Message: "Out of gas, refill pending. Try again later.", Message: "Out of gas, refill pending. Try again later.",
}) })
} }
trackingId := uuid.NewString()
taskPayload, err := json.Marshal(task.TransferPayload{ taskPayload, err := json.Marshal(task.TransferPayload{
TrackingId: trackingId, TrackingId: trackingId,
From: req.From, From: req.From,
@ -84,11 +106,6 @@ func HandleSignTransfer(cu *custodial.Custodial) func(echo.Context) error {
return err return err
} }
err = cu.PgStore.DecrGasQuota(c.Request().Context(), req.From)
if err != nil {
return err
}
return c.JSON(http.StatusOK, OkResp{ return c.JSON(http.StatusOK, OkResp{
Ok: true, Ok: true,
Result: H{ Result: H{

View File

@ -0,0 +1,27 @@
package custodial
import "github.com/grassrootseconomics/w3-celo-patch"
const (
Check = "check"
GiveTo = "giveTo"
MintTo = "mintTo"
NextTime = "nextTime"
Register = "register"
Transfer = "transfer"
TransferFrom = "transferFrom"
)
// 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{
Check: w3.MustNewFunc("check(address)", "bool"),
GiveTo: w3.MustNewFunc("giveTo(address)", "uint256"),
MintTo: w3.MustNewFunc("mintTo(address, uint256)", "bool"),
NextTime: w3.MustNewFunc("nextTime(address)", "uint256"),
Register: w3.MustNewFunc("register(address)", ""),
Transfer: w3.MustNewFunc("transfer(address,uint256)", "bool"),
TransferFrom: w3.MustNewFunc("transferFrom(address, address, uint256)", "bool"),
}
}

View File

@ -1,12 +1,13 @@
package custodial package custodial
import ( import (
"context"
"crypto/ecdsa" "crypto/ecdsa"
"math/big"
"time" "time"
"github.com/bsm/redislock" "github.com/bsm/redislock"
"github.com/celo-org/celo-blockchain/common" "github.com/celo-org/celo-blockchain/common"
eth_crypto "github.com/celo-org/celo-blockchain/crypto"
"github.com/grassrootseconomics/celoutils" "github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/keystore" "github.com/grassrootseconomics/cic-custodial/internal/keystore"
"github.com/grassrootseconomics/cic-custodial/internal/nonce" "github.com/grassrootseconomics/cic-custodial/internal/nonce"
@ -14,26 +15,13 @@ import (
"github.com/grassrootseconomics/cic-custodial/internal/store" "github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/grassrootseconomics/cic-custodial/internal/tasker" "github.com/grassrootseconomics/cic-custodial/internal/tasker"
"github.com/grassrootseconomics/w3-celo-patch" "github.com/grassrootseconomics/w3-celo-patch"
"github.com/grassrootseconomics/w3-celo-patch/module/eth"
"github.com/labstack/gommon/log"
"github.com/redis/go-redis/v9" "github.com/redis/go-redis/v9"
) )
type ( type (
SystemContainer struct { Opts struct {
Abis map[string]*w3.Func
AccountIndexContract common.Address
GasFaucetContract common.Address
GasRefillThreshold *big.Int
GasRefillValue *big.Int
GiftableGasValue *big.Int
GiftableToken common.Address
GiftableTokenValue *big.Int
LockTimeout time.Duration
PrivateKey *ecdsa.PrivateKey
PublicKey string
TokenDecimals int
TokenTransferGasLimit uint64
}
Custodial struct {
CeloProvider *celoutils.Provider CeloProvider *celoutils.Provider
Keystore keystore.Keystore Keystore keystore.Keystore
LockProvider *redislock.Client LockProvider *redislock.Client
@ -41,7 +29,76 @@ type (
PgStore store.Store PgStore store.Store
Pub *pub.Pub Pub *pub.Pub
RedisClient *redis.Client RedisClient *redis.Client
SystemContainer *SystemContainer RegistryAddress string
SystemPrivateKey string
SystemPublicKey string
TaskerClient *tasker.TaskerClient
}
Custodial struct {
Abis map[string]*w3.Func
CeloProvider *celoutils.Provider
Keystore keystore.Keystore
LockProvider *redislock.Client
Noncestore nonce.Noncestore
PgStore store.Store
Pub *pub.Pub
RedisClient *redis.Client
RegistryMap map[string]common.Address
SystemPrivateKey *ecdsa.PrivateKey
SystemPublicKey string
TaskerClient *tasker.TaskerClient TaskerClient *tasker.TaskerClient
} }
) )
func NewCustodial(o Opts) (*Custodial, error) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
registryMap, err := o.CeloProvider.RegistryMap(ctx, celoutils.HexToAddress(o.RegistryAddress))
if err != nil {
log.Errorf("err: %v", err)
return nil, err
}
_, err = o.Noncestore.Peek(ctx, o.SystemPublicKey)
if err == redis.Nil {
// TODO: Bootsrap from Postgres first
var networkNonce uint64
err := o.CeloProvider.Client.CallCtx(
ctx,
eth.Nonce(celoutils.HexToAddress(o.SystemPublicKey), nil).Returns(&networkNonce),
)
if err != nil {
return nil, err
}
if err := o.Noncestore.SetAccountNonce(ctx, o.SystemPublicKey, networkNonce); err != nil {
return nil, err
}
} else if err != nil {
return nil, err
}
privateKey, err := eth_crypto.HexToECDSA(o.SystemPrivateKey)
if err != nil {
return nil, err
}
return &Custodial{
Abis: initAbis(),
CeloProvider: o.CeloProvider,
Keystore: o.Keystore,
LockProvider: o.LockProvider,
Noncestore: o.Noncestore,
PgStore: o.PgStore,
Pub: o.Pub,
RedisClient: o.RedisClient,
RegistryMap: registryMap,
SystemPrivateKey: privateKey,
SystemPublicKey: o.SystemPublicKey,
TaskerClient: o.TaskerClient,
}, nil
}

View File

@ -6,14 +6,16 @@ import (
redispool "github.com/grassrootseconomics/cic-custodial/pkg/redis" redispool "github.com/grassrootseconomics/cic-custodial/pkg/redis"
) )
type Opts struct { type (
Opts struct {
RedisPool *redispool.RedisPool RedisPool *redispool.RedisPool
} }
// RedisNoncestore implements `Noncestore` // RedisNoncestore implements `Noncestore`
type RedisNoncestore struct { RedisNoncestore struct {
redis *redispool.RedisPool redis *redispool.RedisPool
} }
)
func NewRedisNoncestore(o Opts) Noncestore { func NewRedisNoncestore(o Opts) Noncestore {
return &RedisNoncestore{ return &RedisNoncestore{

View File

@ -10,14 +10,8 @@ import (
const ( const (
streamName string = "CUSTODIAL" streamName string = "CUSTODIAL"
streamSubjects string = "CUSTODIAL.*" streamSubjects string = "CUSTODIAL.*"
AccountNewNonce string = "CUSTODIAL.accountNewNonce" AccountActivated string = "CUSTODIAL.accountActivated"
AccountRegister string = "CUSTODIAL.accountRegister" GasRefilled string = "CUSTODIAL.gasRefilled"
AccountGiftGas string = "CUSTODIAL.systemNewAccountGas"
AccountGiftVoucher string = "CUSTODIAL.systemNewAccountVoucher"
AccountRefillGas string = "CUSTODIAL.systemRefillAccountGas"
DispatchFail string = "CUSTODIAL.dispatchFail"
DispatchSuccess string = "CUSTODIAL.dispatchSuccess"
SignTransfer string = "CUSTODIAL.signTransfer"
) )
type ( type (
@ -32,8 +26,6 @@ type (
} }
EventPayload struct { EventPayload struct {
OtxId uint `json:"otxId"`
TrackingId string `json:"trackingId"`
TxHash string `json:"txHash"` TxHash string `json:"txHash"`
} }
) )

View File

@ -4,6 +4,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/store" "github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
) )
@ -22,10 +23,42 @@ func (s *Sub) handler(ctx context.Context, msg *nats.Msg) error {
} }
switch msg.Subject { switch msg.Subject {
case "CHAIN.register":
if chainEvent.Success {
if err := s.cu.PgStore.ActivateAccount(ctx, chainEvent.To); err != nil {
return err
}
eventPayload := &pub.EventPayload{
TxHash: chainEvent.TxHash,
}
if err := s.cu.Pub.Publish(
pub.AccountActivated,
chainEvent.TxHash,
eventPayload,
); err != nil {
return err
}
}
case "CHAIN.gas": case "CHAIN.gas":
if chainEvent.Success {
if err := s.cu.PgStore.ResetGasQuota(ctx, chainEvent.To); err != nil { if err := s.cu.PgStore.ResetGasQuota(ctx, chainEvent.To); err != nil {
return err return err
} }
eventPayload := &pub.EventPayload{
TxHash: chainEvent.TxHash,
}
if err := s.cu.Pub.Publish(
pub.GasRefilled,
chainEvent.TxHash,
eventPayload,
); err != nil {
return err
}
}
} }
return nil return nil

View File

@ -1,45 +0,0 @@
package task
import (
"context"
"encoding/json"
"errors"
"github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/hibiken/asynq"
)
const (
requiredQuorum = 3
)
var (
ErrQuorumNotReached = errors.New("Account activation quorum not reached.")
)
func AccountActivateProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
return func(ctx context.Context, t *asynq.Task) error {
var (
payload AccountPayload
)
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return err
}
quorum, err := cu.PgStore.GetAccountActivationQuorum(ctx, payload.TrackingId)
if err != nil {
return err
}
if quorum < requiredQuorum {
return ErrQuorumNotReached
}
if err := cu.PgStore.ActivateAccount(ctx, payload.PublicKey); err != nil {
return err
}
return nil
}
}

View File

@ -1,145 +0,0 @@
package task
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/bsm/redislock"
"github.com/celo-org/celo-blockchain/common/hexutil"
"github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/grassrootseconomics/cic-custodial/internal/tasker"
"github.com/grassrootseconomics/cic-custodial/pkg/enum"
"github.com/grassrootseconomics/w3-celo-patch"
"github.com/hibiken/asynq"
)
const (
accountActivationCheckDelay = 5 * time.Second
)
func AccountGiftGasProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
return func(ctx context.Context, t *asynq.Task) error {
var (
err error
payload AccountPayload
)
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return fmt.Errorf("account: failed %v: %w", err, asynq.SkipRetry)
}
lock, err := cu.LockProvider.Obtain(
ctx,
lockPrefix+cu.SystemContainer.PublicKey,
cu.SystemContainer.LockTimeout,
&redislock.Options{
RetryStrategy: lockRetry(),
},
)
if err != nil {
return err
}
defer lock.Release(ctx)
nonce, err := cu.Noncestore.Acquire(ctx, cu.SystemContainer.PublicKey)
if err != nil {
return err
}
defer func() {
if err != nil {
if nErr := cu.Noncestore.Return(ctx, cu.SystemContainer.PublicKey); nErr != nil {
err = nErr
}
}
}()
builtTx, err := cu.CeloProvider.SignGasTransferTx(
cu.SystemContainer.PrivateKey,
celoutils.GasTransferTxOpts{
To: w3.A(payload.PublicKey),
Nonce: nonce,
Value: cu.SystemContainer.GiftableGasValue,
GasFeeCap: celoutils.SafeGasFeeCap,
GasTipCap: celoutils.SafeGasTipCap,
},
)
if err != nil {
return err
}
rawTx, err := builtTx.MarshalBinary()
if err != nil {
return err
}
id, err := cu.PgStore.CreateOtx(ctx, store.OTX{
TrackingId: payload.TrackingId,
Type: enum.GIFT_GAS,
RawTx: hexutil.Encode(rawTx),
TxHash: builtTx.Hash().Hex(),
From: cu.SystemContainer.PublicKey,
Data: hexutil.Encode(builtTx.Data()),
GasPrice: builtTx.GasPrice().Uint64(),
GasLimit: builtTx.Gas(),
TransferValue: cu.SystemContainer.GiftableGasValue.Uint64(),
Nonce: builtTx.Nonce(),
})
if err != nil {
return err
}
disptachJobPayload, err := json.Marshal(TxPayload{
OtxId: id,
Tx: builtTx,
})
if err != nil {
return err
}
_, err = cu.TaskerClient.CreateTask(
ctx,
tasker.DispatchTxTask,
tasker.HighPriority,
&tasker.Task{
Payload: disptachJobPayload,
},
)
if err != nil {
return err
}
_, err = cu.TaskerClient.CreateTask(
ctx,
tasker.AccountActivateTask,
tasker.DefaultPriority,
&tasker.Task{
Payload: t.Payload(),
},
asynq.ProcessIn(accountActivationCheckDelay),
)
if err != nil {
return err
}
eventPayload := &pub.EventPayload{
OtxId: id,
TrackingId: payload.TrackingId,
TxHash: builtTx.Hash().Hex(),
}
if err := cu.Pub.Publish(
pub.AccountGiftGas,
builtTx.Hash().Hex(),
eventPayload,
); err != nil {
return err
}
return nil
}
}

View File

@ -1,137 +0,0 @@
package task
import (
"context"
"encoding/json"
"github.com/bsm/redislock"
"github.com/celo-org/celo-blockchain/common/hexutil"
"github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/grassrootseconomics/cic-custodial/internal/tasker"
"github.com/grassrootseconomics/cic-custodial/pkg/enum"
"github.com/grassrootseconomics/w3-celo-patch"
"github.com/hibiken/asynq"
)
func GiftVoucherProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
return func(ctx context.Context, t *asynq.Task) error {
var (
err error
payload AccountPayload
)
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return err
}
lock, err := cu.LockProvider.Obtain(
ctx,
lockPrefix+cu.SystemContainer.PublicKey,
cu.SystemContainer.LockTimeout,
&redislock.Options{
RetryStrategy: lockRetry(),
},
)
if err != nil {
return err
}
defer lock.Release(ctx)
nonce, err := cu.Noncestore.Acquire(ctx, cu.SystemContainer.PublicKey)
if err != nil {
return err
}
defer func() {
if err != nil {
if nErr := cu.Noncestore.Return(ctx, cu.SystemContainer.PublicKey); nErr != nil {
err = nErr
}
}
}()
input, err := cu.SystemContainer.Abis["mintTo"].EncodeArgs(
w3.A(payload.PublicKey),
cu.SystemContainer.GiftableTokenValue,
)
if err != nil {
return err
}
builtTx, err := cu.CeloProvider.SignContractExecutionTx(
cu.SystemContainer.PrivateKey,
celoutils.ContractExecutionTxOpts{
ContractAddress: cu.SystemContainer.GiftableToken,
InputData: input,
GasFeeCap: celoutils.SafeGasFeeCap,
GasTipCap: celoutils.SafeGasTipCap,
GasLimit: cu.SystemContainer.TokenTransferGasLimit,
Nonce: nonce,
},
)
if err != nil {
return err
}
rawTx, err := builtTx.MarshalBinary()
if err != nil {
return err
}
id, err := cu.PgStore.CreateOtx(ctx, store.OTX{
TrackingId: payload.TrackingId,
Type: enum.GIFT_VOUCHER,
RawTx: hexutil.Encode(rawTx),
TxHash: builtTx.Hash().Hex(),
From: cu.SystemContainer.PublicKey,
Data: hexutil.Encode(builtTx.Data()),
GasPrice: builtTx.GasPrice().Uint64(),
GasLimit: builtTx.Gas(),
TransferValue: cu.SystemContainer.GiftableTokenValue.Uint64(),
Nonce: builtTx.Nonce(),
})
if err != nil {
return err
}
disptachJobPayload, err := json.Marshal(TxPayload{
OtxId: id,
Tx: builtTx,
})
if err != nil {
return err
}
_, err = cu.TaskerClient.CreateTask(
ctx,
tasker.DispatchTxTask,
tasker.HighPriority,
&tasker.Task{
Payload: disptachJobPayload,
},
)
if err != nil {
return err
}
eventPayload := &pub.EventPayload{
OtxId: id,
TrackingId: payload.TrackingId,
TxHash: builtTx.Hash().Hex(),
}
if err := cu.Pub.Publish(
pub.AccountGiftVoucher,
builtTx.Hash().Hex(),
eventPayload,
); err != nil {
return err
}
return nil
}
}

View File

@ -1,81 +0,0 @@
package task
import (
"context"
"encoding/json"
"fmt"
"github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/tasker"
"github.com/hibiken/asynq"
)
type AccountPayload struct {
PublicKey string `json:"publicKey"`
TrackingId string `json:"trackingId"`
}
func AccountPrepare(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
return func(ctx context.Context, t *asynq.Task) error {
var payload AccountPayload
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return fmt.Errorf("account: failed %v: %w", err, asynq.SkipRetry)
}
if err := cu.Noncestore.SetAccountNonce(ctx, payload.PublicKey, 0); err != nil {
return err
}
_, err := cu.TaskerClient.CreateTask(
ctx,
tasker.AccountRegisterTask,
tasker.DefaultPriority,
&tasker.Task{
Payload: t.Payload(),
},
)
if err != nil {
return err
}
_, err = cu.TaskerClient.CreateTask(
ctx,
tasker.AccountGiftGasTask,
tasker.DefaultPriority,
&tasker.Task{
Payload: t.Payload(),
},
)
if err != nil {
return err
}
_, err = cu.TaskerClient.CreateTask(
ctx,
tasker.AccountGiftVoucherTask,
tasker.DefaultPriority,
&tasker.Task{
Payload: t.Payload(),
},
)
if err != nil {
return err
}
eventPayload := pub.EventPayload{
TrackingId: payload.TrackingId,
}
if err := cu.Pub.Publish(
pub.AccountNewNonce,
payload.PublicKey,
eventPayload,
); err != nil {
return err
}
return nil
}
}

View File

@ -3,26 +3,22 @@ package task
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "math/big"
"fmt"
"time" "time"
"github.com/bsm/redislock" "github.com/bsm/redislock"
"github.com/celo-org/celo-blockchain/common/hexutil" "github.com/celo-org/celo-blockchain/common/hexutil"
"github.com/grassrootseconomics/celoutils" "github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/custodial" "github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/store" "github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/grassrootseconomics/cic-custodial/internal/tasker" "github.com/grassrootseconomics/cic-custodial/internal/tasker"
"github.com/grassrootseconomics/cic-custodial/pkg/enum" "github.com/grassrootseconomics/cic-custodial/pkg/enum"
"github.com/grassrootseconomics/w3-celo-patch" "github.com/grassrootseconomics/w3-celo-patch/module/eth"
"github.com/hibiken/asynq" "github.com/hibiken/asynq"
"github.com/redis/go-redis/v9"
) )
const ( const (
gasLockPrefix = "gas_lock:" gasGiveToLimit = 250000
gasLockExpiry = 1 * time.Hour
) )
func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error { func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
@ -30,32 +26,75 @@ func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *a
var ( var (
err error err error
payload AccountPayload payload AccountPayload
nextTime big.Int
checkStatus bool
) )
if err := json.Unmarshal(t.Payload(), &payload); err != nil { if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return fmt.Errorf("account: failed %v: %w", err, asynq.SkipRetry) return err
} }
// TODO: Check eth-faucet whether we can request for a topup before signing the tx.
_, gasQuota, err := cu.PgStore.GetAccountStatusByAddress(ctx, payload.PublicKey) _, gasQuota, err := cu.PgStore.GetAccountStatusByAddress(ctx, payload.PublicKey)
if err != nil { if err != nil {
return err return err
} }
gasLock, err := cu.RedisClient.Get(ctx, gasLockPrefix+payload.PublicKey).Bool() // The user has enough gas for atleast 5 more transactions.
if !errors.Is(err, redis.Nil) { if gasQuota > 5 {
return err return nil
} }
if gasQuota > 0 || gasLock { if err := cu.CeloProvider.Client.CallCtx(
ctx,
eth.CallFunc(
cu.Abis[custodial.NextTime],
cu.RegistryMap[celoutils.GasFaucet],
celoutils.HexToAddress(payload.PublicKey),
).Returns(&nextTime),
); err != nil {
return err
}
// The user already requested funds, there is a cooldown applied.
// We can schedule an attempt after the cooldown period has passed.
if nextTime.Int64() > time.Now().Unix() {
_, err = cu.TaskerClient.CreateTask(
ctx,
tasker.AccountRefillGasTask,
tasker.DefaultPriority,
&tasker.Task{
Payload: t.Payload(),
},
asynq.ProcessAt(time.Unix(nextTime.Int64(), 0)),
)
if err != nil {
return err
}
return nil
}
if err := cu.CeloProvider.Client.CallCtx(
ctx,
eth.CallFunc(
cu.Abis[custodial.Check],
cu.RegistryMap[celoutils.GasFaucet],
celoutils.HexToAddress(payload.PublicKey),
).Returns(&checkStatus),
); err != nil {
return err
}
// The gas faucet backend returns a false status, a poke will fail.
if !checkStatus {
return nil return nil
} }
// TODO: Use eth-faucet.
lock, err := cu.LockProvider.Obtain( lock, err := cu.LockProvider.Obtain(
ctx, ctx,
lockPrefix+cu.SystemContainer.PublicKey, lockPrefix+cu.SystemPublicKey,
cu.SystemContainer.LockTimeout, lockTimeout,
&redislock.Options{ &redislock.Options{
RetryStrategy: lockRetry(), RetryStrategy: lockRetry(),
}, },
@ -65,27 +104,33 @@ func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *a
} }
defer lock.Release(ctx) defer lock.Release(ctx)
nonce, err := cu.Noncestore.Acquire(ctx, cu.SystemContainer.PublicKey) nonce, err := cu.Noncestore.Acquire(ctx, cu.SystemPublicKey)
if err != nil { if err != nil {
return err return err
} }
defer func() { defer func() {
if err != nil { if err != nil {
if nErr := cu.Noncestore.Return(ctx, cu.SystemContainer.PublicKey); nErr != nil { if nErr := cu.Noncestore.Return(ctx, cu.SystemPublicKey); nErr != nil {
err = nErr err = nErr
} }
} }
}() }()
// TODO: Review gas params input, err := cu.Abis[custodial.GiveTo].EncodeArgs(
builtTx, err := cu.CeloProvider.SignGasTransferTx( celoutils.HexToAddress(payload.PublicKey),
cu.SystemContainer.PrivateKey, )
celoutils.GasTransferTxOpts{ if err != nil {
To: w3.A(payload.PublicKey), return err
Nonce: nonce, }
Value: cu.SystemContainer.GiftableGasValue,
builtTx, err := cu.CeloProvider.SignContractExecutionTx(
cu.SystemPrivateKey,
celoutils.ContractExecutionTxOpts{
ContractAddress: cu.RegistryMap[celoutils.GasFaucet],
InputData: input,
GasFeeCap: celoutils.SafeGasFeeCap, GasFeeCap: celoutils.SafeGasFeeCap,
GasTipCap: celoutils.SafeGasTipCap, GasTipCap: celoutils.SafeGasTipCap,
Nonce: nonce,
}, },
) )
if err != nil { if err != nil {
@ -102,11 +147,10 @@ func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *a
Type: enum.REFILL_GAS, Type: enum.REFILL_GAS,
RawTx: hexutil.Encode(rawTx), RawTx: hexutil.Encode(rawTx),
TxHash: builtTx.Hash().Hex(), TxHash: builtTx.Hash().Hex(),
From: cu.SystemContainer.PublicKey, From: cu.SystemPublicKey,
Data: hexutil.Encode(builtTx.Data()), Data: hexutil.Encode(builtTx.Data()),
GasPrice: builtTx.GasPrice().Uint64(), GasPrice: builtTx.GasPrice().Uint64(),
GasLimit: builtTx.Gas(), GasLimit: builtTx.Gas(),
TransferValue: cu.SystemContainer.GiftableGasValue.Uint64(),
Nonce: builtTx.Nonce(), Nonce: builtTx.Nonce(),
}) })
if err != nil { if err != nil {
@ -133,24 +177,6 @@ func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *a
return err return err
} }
eventPayload := &pub.EventPayload{
OtxId: id,
TrackingId: payload.TrackingId,
TxHash: builtTx.Hash().Hex(),
}
if err := cu.Pub.Publish(
pub.AccountRefillGas,
builtTx.Hash().Hex(),
eventPayload,
); err != nil {
return err
}
if _, err := cu.RedisClient.SetEx(ctx, gasLockPrefix+payload.PublicKey, true, gasLockExpiry).Result(); err != nil {
return err
}
return nil return nil
} }
} }

View File

@ -3,20 +3,22 @@ package task
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"github.com/bsm/redislock" "github.com/bsm/redislock"
"github.com/celo-org/celo-blockchain/common/hexutil" "github.com/celo-org/celo-blockchain/common/hexutil"
"github.com/grassrootseconomics/celoutils" "github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/custodial" "github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/store" "github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/grassrootseconomics/cic-custodial/internal/tasker" "github.com/grassrootseconomics/cic-custodial/internal/tasker"
"github.com/grassrootseconomics/cic-custodial/pkg/enum" "github.com/grassrootseconomics/cic-custodial/pkg/enum"
"github.com/grassrootseconomics/w3-celo-patch"
"github.com/hibiken/asynq" "github.com/hibiken/asynq"
) )
type AccountPayload struct {
PublicKey string `json:"publicKey"`
TrackingId string `json:"trackingId"`
}
func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error { func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
return func(ctx context.Context, t *asynq.Task) error { return func(ctx context.Context, t *asynq.Task) error {
var ( var (
@ -25,13 +27,13 @@ func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Conte
) )
if err := json.Unmarshal(t.Payload(), &payload); err != nil { if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return fmt.Errorf("account: failed %v: %w", err, asynq.SkipRetry) return err
} }
lock, err := cu.LockProvider.Obtain( lock, err := cu.LockProvider.Obtain(
ctx, ctx,
lockPrefix+cu.SystemContainer.PublicKey, lockPrefix+cu.SystemPublicKey,
cu.SystemContainer.LockTimeout, lockTimeout,
&redislock.Options{ &redislock.Options{
RetryStrategy: lockRetry(), RetryStrategy: lockRetry(),
}, },
@ -41,34 +43,33 @@ func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Conte
} }
defer lock.Release(ctx) defer lock.Release(ctx)
nonce, err := cu.Noncestore.Acquire(ctx, cu.SystemContainer.PublicKey) nonce, err := cu.Noncestore.Acquire(ctx, cu.SystemPublicKey)
if err != nil { if err != nil {
return err return err
} }
defer func() { defer func() {
if err != nil { if err != nil {
if nErr := cu.Noncestore.Return(ctx, cu.SystemContainer.PublicKey); nErr != nil { if nErr := cu.Noncestore.Return(ctx, cu.SystemPublicKey); nErr != nil {
err = nErr err = nErr
} }
} }
}() }()
input, err := cu.SystemContainer.Abis["add"].EncodeArgs( input, err := cu.Abis[custodial.Register].EncodeArgs(
w3.A(payload.PublicKey), celoutils.HexToAddress(payload.PublicKey),
) )
if err != nil { if err != nil {
return err return err
} }
// TODO: Review gas params.
builtTx, err := cu.CeloProvider.SignContractExecutionTx( builtTx, err := cu.CeloProvider.SignContractExecutionTx(
cu.SystemContainer.PrivateKey, cu.SystemPrivateKey,
celoutils.ContractExecutionTxOpts{ celoutils.ContractExecutionTxOpts{
ContractAddress: cu.SystemContainer.AccountIndexContract, ContractAddress: cu.RegistryMap[celoutils.CustodialProxy],
InputData: input, InputData: input,
GasFeeCap: celoutils.SafeGasFeeCap, GasFeeCap: celoutils.SafeGasFeeCap,
GasTipCap: celoutils.SafeGasTipCap, GasTipCap: celoutils.SafeGasTipCap,
GasLimit: cu.SystemContainer.TokenTransferGasLimit, GasLimit: gasLimit,
Nonce: nonce, Nonce: nonce,
}, },
) )
@ -86,7 +87,7 @@ func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Conte
Type: enum.ACCOUNT_REGISTER, Type: enum.ACCOUNT_REGISTER,
RawTx: hexutil.Encode(rawTx), RawTx: hexutil.Encode(rawTx),
TxHash: builtTx.Hash().Hex(), TxHash: builtTx.Hash().Hex(),
From: cu.SystemContainer.PublicKey, From: cu.SystemPublicKey,
Data: hexutil.Encode(builtTx.Data()), Data: hexutil.Encode(builtTx.Data()),
GasPrice: builtTx.GasPrice().Uint64(), GasPrice: builtTx.GasPrice().Uint64(),
GasLimit: builtTx.Gas(), GasLimit: builtTx.Gas(),
@ -116,17 +117,7 @@ func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Conte
return err return err
} }
eventPayload := &pub.EventPayload{ if err := cu.Noncestore.SetAccountNonce(ctx, payload.PublicKey, 0); err != nil {
OtxId: id,
TrackingId: payload.TrackingId,
TxHash: builtTx.Hash().Hex(),
}
if err := cu.Pub.Publish(
pub.AccountRegister,
builtTx.Hash().Hex(),
eventPayload,
); err != nil {
return err return err
} }

View File

@ -9,7 +9,6 @@ import (
"github.com/celo-org/celo-blockchain/core/types" "github.com/celo-org/celo-blockchain/core/types"
"github.com/grassrootseconomics/celoutils" "github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/custodial" "github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/store" "github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/grassrootseconomics/cic-custodial/pkg/enum" "github.com/grassrootseconomics/cic-custodial/pkg/enum"
"github.com/grassrootseconomics/w3-celo-patch/module/eth" "github.com/grassrootseconomics/w3-celo-patch/module/eth"
@ -26,18 +25,14 @@ func DispatchTx(cu *custodial.Custodial) func(context.Context, *asynq.Task) erro
var ( var (
payload TxPayload payload TxPayload
dispatchStatus store.DispatchStatus dispatchStatus store.DispatchStatus
eventPayload pub.EventPayload
dispathchTx common.Hash dispathchTx common.Hash
) )
if err := json.Unmarshal(t.Payload(), &payload); err != nil { if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry) return err
} }
txHash := payload.Tx.Hash().Hex() dispatchStatus.OtxId = payload.OtxId
dispatchStatus.OtxId, eventPayload.OtxId = payload.OtxId, payload.OtxId
eventPayload.TxHash = txHash
if err := cu.CeloProvider.Client.CallCtx( if err := cu.CeloProvider.Client.CallCtx(
ctx, ctx,
@ -58,10 +53,6 @@ func DispatchTx(cu *custodial.Custodial) func(context.Context, *asynq.Task) erro
return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry) return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry)
} }
if err := cu.Pub.Publish(pub.DispatchFail, txHash, eventPayload); err != nil {
return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry)
}
return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry) return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry)
} }
@ -71,10 +62,6 @@ func DispatchTx(cu *custodial.Custodial) func(context.Context, *asynq.Task) erro
return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry) return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry)
} }
if err := cu.Pub.Publish(pub.DispatchSuccess, txHash, eventPayload); err != nil {
return fmt.Errorf("dispatch: failed %v: %w", err, asynq.SkipRetry)
}
return nil return nil
} }
} }

View File

@ -3,23 +3,19 @@ package task
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"math/big" "math/big"
"github.com/bsm/redislock" "github.com/bsm/redislock"
"github.com/celo-org/celo-blockchain/common/hexutil" "github.com/celo-org/celo-blockchain/common/hexutil"
"github.com/grassrootseconomics/celoutils" "github.com/grassrootseconomics/celoutils"
"github.com/grassrootseconomics/cic-custodial/internal/custodial" "github.com/grassrootseconomics/cic-custodial/internal/custodial"
"github.com/grassrootseconomics/cic-custodial/internal/pub"
"github.com/grassrootseconomics/cic-custodial/internal/store" "github.com/grassrootseconomics/cic-custodial/internal/store"
"github.com/grassrootseconomics/cic-custodial/internal/tasker" "github.com/grassrootseconomics/cic-custodial/internal/tasker"
"github.com/grassrootseconomics/cic-custodial/pkg/enum" "github.com/grassrootseconomics/cic-custodial/pkg/enum"
"github.com/grassrootseconomics/w3-celo-patch"
"github.com/hibiken/asynq" "github.com/hibiken/asynq"
) )
type ( type TransferPayload struct {
TransferPayload struct {
TrackingId string `json:"trackingId"` TrackingId string `json:"trackingId"`
From string `json:"from" ` From string `json:"from" `
To string `json:"to"` To string `json:"to"`
@ -27,14 +23,6 @@ type (
Amount uint64 `json:"amount"` Amount uint64 `json:"amount"`
} }
transferEventPayload struct {
DispatchTaskId string `json:"dispatchTaskId"`
OTXId uint `json:"otxId"`
TrackingId string `json:"trackingId"`
TxHash string `json:"txHash"`
}
)
func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) error { func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
return func(ctx context.Context, t *asynq.Task) error { return func(ctx context.Context, t *asynq.Task) error {
var ( var (
@ -43,13 +31,13 @@ func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) er
) )
if err := json.Unmarshal(t.Payload(), &payload); err != nil { if err := json.Unmarshal(t.Payload(), &payload); err != nil {
return fmt.Errorf("account: failed %v: %w", err, asynq.SkipRetry) return err
} }
lock, err := cu.LockProvider.Obtain( lock, err := cu.LockProvider.Obtain(
ctx, ctx,
lockPrefix+payload.From, lockPrefix+payload.From,
cu.SystemContainer.LockTimeout, lockTimeout,
&redislock.Options{ &redislock.Options{
RetryStrategy: lockRetry(), RetryStrategy: lockRetry(),
}, },
@ -70,26 +58,25 @@ func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) er
} }
defer func() { defer func() {
if err != nil { if err != nil {
if nErr := cu.Noncestore.Return(ctx, cu.SystemContainer.PublicKey); nErr != nil { if nErr := cu.Noncestore.Return(ctx, cu.SystemPublicKey); nErr != nil {
err = nErr err = nErr
} }
} }
}() }()
input, err := cu.SystemContainer.Abis["transfer"].EncodeArgs(w3.A(payload.To), new(big.Int).SetUint64(payload.Amount)) input, err := cu.Abis[custodial.Transfer].EncodeArgs(celoutils.HexToAddress(payload.To), new(big.Int).SetUint64(payload.Amount))
if err != nil { if err != nil {
return err return err
} }
// TODO: Review gas params.
builtTx, err := cu.CeloProvider.SignContractExecutionTx( builtTx, err := cu.CeloProvider.SignContractExecutionTx(
key, key,
celoutils.ContractExecutionTxOpts{ celoutils.ContractExecutionTxOpts{
ContractAddress: w3.A(payload.VoucherAddress), ContractAddress: celoutils.HexToAddress(payload.VoucherAddress),
InputData: input, InputData: input,
GasFeeCap: celoutils.SafeGasFeeCap, GasFeeCap: celoutils.SafeGasFeeCap,
GasTipCap: celoutils.SafeGasTipCap, GasTipCap: celoutils.SafeGasTipCap,
GasLimit: cu.SystemContainer.TokenTransferGasLimit, GasLimit: gasLimit,
Nonce: nonce, Nonce: nonce,
}, },
) )
@ -118,6 +105,10 @@ func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) er
return err return err
} }
if err := cu.PgStore.DecrGasQuota(ctx, payload.From); err != nil {
return err
}
disptachJobPayload, err := json.Marshal(TxPayload{ disptachJobPayload, err := json.Marshal(TxPayload{
OtxId: id, OtxId: id,
Tx: builtTx, Tx: builtTx,
@ -158,20 +149,6 @@ func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) er
return err return err
} }
eventPayload := &transferEventPayload{
OTXId: id,
TrackingId: payload.TrackingId,
TxHash: builtTx.Hash().Hex(),
}
if err := cu.Pub.Publish(
pub.SignTransfer,
builtTx.Hash().Hex(),
eventPayload,
); err != nil {
return err
}
return nil return nil
} }
} }

View File

@ -7,8 +7,11 @@ import (
) )
const ( const (
gasLimit = 250000
lockPrefix = "lock:" lockPrefix = "lock:"
lockRetryDelay = 25 * time.Millisecond lockRetryDelay = 25 * time.Millisecond
lockTimeout = 1 * time.Second
) )
// lockRetry will at most try to obtain the lock 20 times within ~0.5s. // lockRetry will at most try to obtain the lock 20 times within ~0.5s.

View File

@ -15,12 +15,8 @@ type Task struct {
} }
const ( const (
AccountPrepareTask TaskName = "sys:prepare_account"
AccountRegisterTask TaskName = "sys:register_account" AccountRegisterTask TaskName = "sys:register_account"
AccountGiftGasTask TaskName = "sys:gift_gas"
AccountGiftVoucherTask TaskName = "sys:gift_token"
AccountRefillGasTask TaskName = "sys:refill_gas" AccountRefillGasTask TaskName = "sys:refill_gas"
AccountActivateTask TaskName = "sys:quorum_check"
SignTransferTask TaskName = "usr:sign_transfer" SignTransferTask TaskName = "usr:sign_transfer"
DispatchTxTask TaskName = "rpc:dispatch" DispatchTxTask TaskName = "rpc:dispatch"
) )

View File

@ -19,9 +19,7 @@ const (
FAIL_UNKNOWN_RPC_ERROR OtxStatus = "FAIL_UNKNOWN_RPC_ERROR" FAIL_UNKNOWN_RPC_ERROR OtxStatus = "FAIL_UNKNOWN_RPC_ERROR"
REVERTED OtxStatus = "REVERTED" REVERTED OtxStatus = "REVERTED"
GIFT_GAS OtxType = "GIFT_GAS"
ACCOUNT_REGISTER OtxType = "ACCOUNT_REGISTER" ACCOUNT_REGISTER OtxType = "ACCOUNT_REGISTER"
GIFT_VOUCHER OtxType = "GIFT_VOUCHER"
REFILL_GAS OtxType = "REFILL_GAS" REFILL_GAS OtxType = "REFILL_GAS"
TRANSFER_VOUCHER OtxType = "TRANSFER_VOUCHER" TRANSFER_VOUCHER OtxType = "TRANSFER_VOUCHER"
) )