wip: refactor taskers

internal/tasker/task/account.go

@@ -2,6 +2,7 @@ package task
 
 import (
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"math/big"
@@ -9,27 +10,40 @@ import (
 	"github.com/bsm/redislock"
 	celo "github.com/grassrootseconomics/cic-celo-sdk"
 	"github.com/grassrootseconomics/cic-custodial/internal/nonce"
+	"github.com/grassrootseconomics/cic-custodial/internal/store"
 	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
 	"github.com/grassrootseconomics/w3-celo-patch"
 	"github.com/grassrootseconomics/w3-celo-patch/module/eth"
 	"github.com/hibiken/asynq"
+	"github.com/nats-io/nats.go"
 )
 
-type SystemPayload struct {
-	PublicKey string `json:"publicKey"`
-}
+type (
+	AccountPayload struct {
+		PublicKey  string `json:"publicKey"`
+		TrackingId string `json:"trackingId"`
+	}
+
+	accountEventPayload struct {
+		TrackingId string `json:"trackingId"`
+	}
+)
 
 func PrepareAccount(
-	nonceProvider nonce.Noncestore,
+	js nats.JetStreamContext,
+	noncestore nonce.Noncestore,
 	taskerClient *tasker.TaskerClient,
 ) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
-		var p SystemPayload
+		var (
+			p AccountPayload
+		)
+
 		if err := json.Unmarshal(t.Payload(), &p); err != nil {
-			return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
+			return err
 		}
 
-		if err := nonceProvider.SetNewAccountNonce(ctx, p.PublicKey); err != nil {
+		if err := noncestore.SetNewAccountNonce(ctx, p.PublicKey); err != nil {
 			return err
 		}
 
@@ -55,21 +69,40 @@ func PrepareAccount(
 			return err
 		}
 
+		eventPayload := &accountEventPayload{
+			TrackingId: p.TrackingId,
+		}
+
+		eventJson, err := json.Marshal(eventPayload)
+		if err != nil {
+			return err
+		}
+
+		_, err = js.Publish("CUSTODIAL.accountNewNonce", eventJson)
+		if err != nil {
+			return err
+		}
+
 		return nil
 	}
 }
 
 func GiftGasProcessor(
 	celoProvider *celo.Provider,
-	nonceProvider nonce.Noncestore,
+	js nats.JetStreamContext,
 	lockProvider *redislock.Client,
+	noncestore nonce.Noncestore,
+	pg store.Store,
 	system *tasker.SystemContainer,
 	taskerClient *tasker.TaskerClient,
 ) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
-		var p SystemPayload
+		var (
+			p AccountPayload
+		)
+
 		if err := json.Unmarshal(t.Payload(), &p); err != nil {
-			return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
+			return err
 		}
 
 		lock, err := lockProvider.Obtain(ctx, system.LockPrefix+system.PublicKey, system.LockTimeout, nil)
@@ -78,11 +111,12 @@ func GiftGasProcessor(
 		}
 		defer lock.Release(ctx)
 
-		nonce, err := nonceProvider.Acquire(ctx, system.PublicKey)
+		nonce, err := noncestore.Acquire(ctx, system.PublicKey)
 		if err != nil {
 			return err
 		}
 
+		// TODO: Review gas params
 		builtTx, err := celoProvider.SignGasTransferTx(
 			system.PrivateKey,
 			celo.GasTransferTxOpts{
@@ -93,17 +127,49 @@ func GiftGasProcessor(
 			},
 		)
 		if err != nil {
-			if err := nonceProvider.Return(ctx, p.PublicKey); err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
 				return err
 			}
-			return fmt.Errorf("nonce.Return failed: %v: %w", err, asynq.SkipRetry)
+
+			return err
+		}
+
+		rawTx, err := builtTx.MarshalBinary()
+		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		id, err := pg.CreateOTX(ctx, store.OTX{
+			RawTx:    hex.EncodeToString(rawTx),
+			TxHash:   builtTx.Hash().Hex(),
+			From:     system.PublicKey,
+			Data:     string(builtTx.Data()),
+			GasPrice: builtTx.GasPrice().Uint64(),
+			Nonce:    builtTx.Nonce(),
+		})
+		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
 		}
 
 		disptachJobPayload, err := json.Marshal(TxPayload{
-			Tx: builtTx,
+			OtxId:      id,
+			TrackingId: p.TrackingId,
+			Tx:         builtTx,
 		})
 		if err != nil {
-			return fmt.Errorf("json.Marshal failed: %v: %w", err, asynq.SkipRetry)
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
 		}
 
 		_, err = taskerClient.CreateTask(
@@ -114,6 +180,28 @@ func GiftGasProcessor(
 			},
 		)
 		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		eventPayload := &accountEventPayload{
+			TrackingId: p.TrackingId,
+		}
+
+		eventJson, err := json.Marshal(eventPayload)
+		if err != nil {
+			return err
+		}
+
+		_, err = js.Publish("CUSTODIAL.giftNewAccountGas", eventJson)
+		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
 			return err
 		}
 
@@ -123,18 +211,21 @@ func GiftGasProcessor(
 
 func GiftTokenProcessor(
 	celoProvider *celo.Provider,
-	nonceProvider nonce.Noncestore,
+	js nats.JetStreamContext,
 	lockProvider *redislock.Client,
+	noncestore nonce.Noncestore,
+	pg store.Store,
 	system *tasker.SystemContainer,
 	taskerClient *tasker.TaskerClient,
 ) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
-		var p SystemPayload
-		if err := json.Unmarshal(t.Payload(), &p); err != nil {
-			return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
-		}
+		var (
+			p AccountPayload
+		)
 
-		publicKey := w3.A(p.PublicKey)
+		if err := json.Unmarshal(t.Payload(), &p); err != nil {
+			return err
+		}
 
 		lock, err := lockProvider.Obtain(ctx, system.LockPrefix+system.PublicKey, system.LockTimeout, nil)
 		if err != nil {
@@ -142,38 +233,70 @@ func GiftTokenProcessor(
 		}
 		defer lock.Release(ctx)
 
-		nonce, err := nonceProvider.Acquire(ctx, system.PublicKey)
+		nonce, err := noncestore.Acquire(ctx, system.PublicKey)
 		if err != nil {
 			return err
 		}
 
-		input, err := system.Abis["mint"].EncodeArgs(publicKey, system.GiftableTokenValue)
+		input, err := system.Abis["mintTo"].EncodeArgs(w3.A(p.PublicKey), system.GiftableTokenValue)
 		if err != nil {
-			return fmt.Errorf("ABI encode failed %v: %w", err, asynq.SkipRetry)
+			return err
 		}
 
+		// TODO: Review gas params.
 		builtTx, err := celoProvider.SignContractExecutionTx(
 			system.PrivateKey,
 			celo.ContractExecutionTxOpts{
 				ContractAddress: system.GiftableToken,
 				InputData:       input,
-				GasPrice:        celo.FixedMinGas,
+				GasPrice:        big.NewInt(20000000000),
 				GasLimit:        system.TokenTransferGasLimit,
 				Nonce:           nonce,
 			},
 		)
 		if err != nil {
-			if err := nonceProvider.Return(ctx, p.PublicKey); err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
 				return err
 			}
-			return fmt.Errorf("nonce.Return failed: %v: %w", err, asynq.SkipRetry)
+			return err
+		}
+
+		rawTx, err := builtTx.MarshalBinary()
+		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		id, err := pg.CreateOTX(ctx, store.OTX{
+			RawTx:    hex.EncodeToString(rawTx),
+			TxHash:   builtTx.Hash().Hex(),
+			From:     system.PublicKey,
+			Data:     string(builtTx.Data()),
+			GasPrice: builtTx.GasPrice().Uint64(),
+			Nonce:    builtTx.Nonce(),
+		})
+		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
 		}
 
 		disptachJobPayload, err := json.Marshal(TxPayload{
-			Tx: builtTx,
+			OtxId:      id,
+			TrackingId: p.TrackingId,
+			Tx:         builtTx,
 		})
 		if err != nil {
-			return fmt.Errorf("json.Marshal failed: %v: %w", err, asynq.SkipRetry)
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
 		}
 
 		_, err = taskerClient.CreateTask(
@@ -184,6 +307,28 @@ func GiftTokenProcessor(
 			},
 		)
 		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		eventPayload := &accountEventPayload{
+			TrackingId: p.TrackingId,
+		}
+
+		eventJson, err := json.Marshal(eventPayload)
+		if err != nil {
+			return err
+		}
+
+		_, err = js.Publish("CUSTODIAL.giftNewAccountVoucher", eventJson)
+		if err != nil {
+			if err := noncestore.Return(ctx, system.PublicKey); err != nil {
+				return err
+			}
+
 			return err
 		}
 
@@ -192,6 +337,7 @@ func GiftTokenProcessor(
 }
 
 // TODO: https://github.com/grassrootseconomics/cic-custodial/issues/43
+// TODO:
 func RefillGasProcessor(
 	celoProvider *celo.Provider,
 	nonceProvider nonce.Noncestore,
@@ -201,7 +347,7 @@ func RefillGasProcessor(
 ) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
 		var (
-			p       SystemPayload
+			p       AccountPayload
 			balance big.Int
 		)
 		if err := json.Unmarshal(t.Payload(), &p); err != nil {

internal/tasker/task/dispatch.go

@@ -3,21 +3,35 @@ package task
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 
 	"github.com/celo-org/celo-blockchain/common"
 	"github.com/celo-org/celo-blockchain/core/types"
 	celo "github.com/grassrootseconomics/cic-celo-sdk"
+	"github.com/grassrootseconomics/cic-custodial/internal/store"
+	"github.com/grassrootseconomics/cic-custodial/pkg/status"
 	"github.com/grassrootseconomics/w3-celo-patch/module/eth"
 	"github.com/hibiken/asynq"
+	"github.com/nats-io/nats.go"
 )
 
-type TxPayload struct {
-	Tx *types.Transaction `json:"tx"`
-}
+type (
+	TxPayload struct {
+		OtxId      uint               `json:"otxId"`
+		TrackingId string             `json:"trackingId"`
+		Tx         *types.Transaction `json:"tx"`
+	}
+
+	dispatchEventPayload struct {
+		TrackingId string
+		TxHash     string
+	}
+)
 
 func TxDispatch(
 	celoProvider *celo.Provider,
+	js nats.JetStreamContext,
+	pg store.Store,
+
 ) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
 		var (
@@ -26,14 +40,53 @@ func TxDispatch(
 		)
 
 		if err := json.Unmarshal(t.Payload(), &p); err != nil {
-			return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
+			return err
+		}
+
+		dispatchStatus := store.DispatchStatus{
+			OtxId:      p.OtxId,
+			TrackingId: p.TrackingId,
+		}
+
+		eventPayload := &dispatchEventPayload{
+			TrackingId: p.TrackingId,
 		}
 
-		// TODO: Handle all fail cases
 		if err := celoProvider.Client.CallCtx(
 			ctx,
 			eth.SendTx(p.Tx).Returns(&txHash),
 		); err != nil {
+			// TODO: Coreect error status
+			dispatchStatus.Status = status.FailGasPrice
+
+			_, err := pg.CreateDispatchStatus(ctx, dispatchStatus)
+			if err != nil {
+				return err
+			}
+
+			eventJson, err := json.Marshal(eventPayload)
+			if err != nil {
+				return err
+			}
+
+			_, err = js.Publish("CUSTODIAL.dispatchFail", eventJson, nats.MsgId(txHash.Hex()))
+			if err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		dispatchStatus.TrackingId = status.Successful
+		eventPayload.TxHash = txHash.Hex()
+
+		eventJson, err := json.Marshal(eventPayload)
+		if err != nil {
+			return err
+		}
+
+		_, err = js.Publish("CUSTODIAL.dispatchSuccessful", eventJson, nats.MsgId(txHash.Hex()))
+		if err != nil {
 			return err
 		}
 

internal/tasker/task/sign.go

@@ -0,0 +1,183 @@
+package task
+
+import (
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"math/big"
+
+	"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/store"
+	"github.com/grassrootseconomics/cic-custodial/internal/tasker"
+	"github.com/grassrootseconomics/w3-celo-patch"
+	"github.com/hibiken/asynq"
+	"github.com/nats-io/nats.go"
+	"github.com/zerodha/logf"
+)
+
+type (
+	TransferPayload struct {
+		TrackingId     string `json:"trackingId"`
+		From           string `json:"from" `
+		To             string `json:"to"`
+		VoucherAddress string `json:"voucherAddress"`
+		Amount         int64  `json:"amount"`
+	}
+
+	transferEventPayload struct {
+		DispatchTaskId string `json:"dispatchTaskId"`
+		OTXId          uint   `json:"otxId"`
+		TrackingId     string `json:"trackingId"`
+		TxHash         string `json:"txHash"`
+	}
+)
+
+func SignTransfer(
+	celoProvider *celo.Provider,
+	js nats.JetStreamContext,
+	keystore keystore.Keystore,
+	lockProvider *redislock.Client,
+	noncestore nonce.Noncestore,
+	pg store.Store,
+	system *tasker.SystemContainer,
+	taskerClient *tasker.TaskerClient,
+	logger logf.Logger,
+) func(context.Context, *asynq.Task) error {
+	return func(ctx context.Context, t *asynq.Task) error {
+		var (
+			p TransferPayload
+		)
+
+		if err := json.Unmarshal(t.Payload(), &p); err != nil {
+			return err
+		}
+
+		lock, err := lockProvider.Obtain(
+			ctx,
+			system.LockPrefix+p.From,
+			system.LockTimeout,
+			nil,
+		)
+		if err != nil {
+			return err
+		}
+		defer lock.Release(ctx)
+
+		key, err := keystore.LoadPrivateKey(ctx, p.From)
+		if err != nil {
+			return err
+		}
+
+		nonce, err := noncestore.Acquire(ctx, p.From)
+		if err != nil {
+			return err
+		}
+
+		input, err := system.Abis["transfer"].EncodeArgs(w3.A(p.To), big.NewInt(p.Amount))
+		if err != nil {
+			return err
+		}
+
+		// TODO: Review gas params.
+		builtTx, err := celoProvider.SignContractExecutionTx(
+			key,
+			celo.ContractExecutionTxOpts{
+				ContractAddress: w3.A(p.VoucherAddress),
+				InputData:       input,
+				GasPrice:        big.NewInt(20000000000),
+				GasLimit:        system.TokenTransferGasLimit,
+				Nonce:           nonce,
+			},
+		)
+		if err != nil {
+			if err := noncestore.Return(ctx, p.From); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		rawTx, err := builtTx.MarshalBinary()
+		if err != nil {
+			if err := noncestore.Return(ctx, p.From); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		id, err := pg.CreateOTX(ctx, store.OTX{
+			RawTx:    hex.EncodeToString(rawTx),
+			TxHash:   builtTx.Hash().Hex(),
+			From:     p.From,
+			Data:     string(builtTx.Data()),
+			GasPrice: builtTx.GasPrice().Uint64(),
+			Nonce:    builtTx.Nonce(),
+		})
+		if err != nil {
+			if err := noncestore.Return(ctx, p.From); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		disptachJobPayload, err := json.Marshal(TxPayload{
+			OtxId:      id,
+			TrackingId: p.TrackingId,
+			Tx:         builtTx,
+		})
+		if err != nil {
+			if err := noncestore.Return(ctx, p.From); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		dispatchTask, err := taskerClient.CreateTask(
+			tasker.TxDispatchTask,
+			tasker.HighPriority,
+			&tasker.Task{
+				Payload: disptachJobPayload,
+			},
+		)
+		if err != nil {
+			if err := noncestore.Return(ctx, p.From); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		eventPayload := &transferEventPayload{
+			DispatchTaskId: dispatchTask.ID,
+			OTXId:          id,
+			TrackingId:     p.TrackingId,
+			TxHash:         builtTx.Hash().Hex(),
+		}
+
+		eventJson, err := json.Marshal(eventPayload)
+		if err != nil {
+			if err := noncestore.Return(ctx, p.From); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		_, err = js.Publish("CUSTODIAL.transferSign", eventJson, nats.MsgId(builtTx.Hash().Hex()))
+		if err != nil {
+			if err := noncestore.Return(ctx, p.From); err != nil {
+				return err
+			}
+
+			return err
+		}
+
+		return nil
+	}
+}