feat (wip): add in-built chain subscription

* BREAKING: remove events interface -> use jetstream struct
* in-built chain subscription to update otx status
* potential fix for tasks: defer rollback nonce error detection

cmd/service/init.go

@@ -48,9 +48,9 @@ func initConfig(configFilePath string) *koanf.Koanf {
 		lo.Fatal("Could not load config file", "error", err)
 	}
 
-	if err := ko.Load(env.Provider("CUSTODIAL_", ".", func(s string) string {
+	if err := ko.Load(env.Provider("", ".", func(s string) string {
-		return strings.Replace(strings.ToLower(
+		return strings.ReplaceAll(strings.ToLower(
-			strings.TrimPrefix(s, "")), "_", ".", -1)
+			strings.TrimPrefix(s, "")), "_", ".")
 	}), nil); err != nil {
 		lo.Fatal("Could not override config from env vars", "error", err)
 	}
@@ -178,7 +178,7 @@ func initPostgresStore(postgresPool *pgxpool.Pool, queries *queries.Queries) sto
 }
 
 // Init JetStream context for tasker events.
-func initJetStream() (events.EventEmitter, error) {
+func initJetStream() (*events.JetStream, error) {
 	jsEmitter, err := events.NewJetStreamEventEmitter(events.JetStreamOpts{
 		ServerUrl:       ko.MustString("jetstream.endpoint"),
 		PersistDuration: time.Duration(ko.MustInt("jetstream.persist_duration_hrs")) * time.Hour,

internal/custodial/custodial.go

@@ -12,7 +12,7 @@ import (
 
 type Custodial struct {
 	CeloProvider    *celoutils.Provider
-	EventEmitter    events.EventEmitter
+	EventEmitter    *events.JetStream
 	Keystore        keystore.Keystore
 	LockProvider    *redislock.Client
 	Noncestore      nonce.Noncestore

internal/events/events.go

@@ -1,10 +1,5 @@
 package events
 
-type EventEmitter interface {
-	Close()
-	Publish(subject string, dedupId string, eventPayload interface{}) error
-}
-
 type EventPayload struct {
 	OtxId      uint   `json:"otxId"`
 	TrackingId string `json:"trackingId"`

internal/events/jetstream.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"github.com/zerodha/logf"
 )
 
 const (
@@ -22,17 +23,19 @@ const (
 )
 
 type JetStreamOpts struct {
+	Logg logf.Logger
 	ServerUrl       string
 	PersistDuration time.Duration
 	DedupDuration   time.Duration
 }
 
 type JetStream struct {
+	logg logf.Logger
 	jsCtx    nats.JetStreamContext
-	nc    *nats.Conn
+	natsConn *nats.Conn
 }
 
-func NewJetStreamEventEmitter(o JetStreamOpts) (EventEmitter, error) {
+func NewJetStreamEventEmitter(o JetStreamOpts) (*JetStream, error) {
 	natsConn, err := nats.Connect(o.ServerUrl)
 	if err != nil {
 		return nil, err
@@ -60,14 +63,14 @@ func NewJetStreamEventEmitter(o JetStreamOpts) (EventEmitter, error) {
 
 	return &JetStream{
 		jsCtx:    js,
-		nc:    natsConn,
+		natsConn: natsConn,
 	}, nil
 }
 
 // Close gracefully shutdowns the JetStream connection.
 func (js *JetStream) Close() {
-	if js.nc != nil {
+	if js.natsConn != nil {
-		js.nc.Close()
+		js.natsConn.Close()
 	}
 }
 

internal/events/jetstream_sub.go

@@ -0,0 +1,54 @@
+package events
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/grassrootseconomics/cic-custodial/internal/store"
+	"github.com/nats-io/nats.go"
+)
+
+const (
+	backOffTimer = 2 * time.Second
+	durableId    = "cic-custodial"
+	pullStream   = "CHAIN"
+	pullSubject  = "CHAIN.*"
+)
+
+func (js *JetStream) ChainSubscription(ctx context.Context, store store.Store) error {
+	subOpts := []nats.SubOpt{
+		nats.ManualAck(),
+		nats.Bind(pullStream, durableId),
+	}
+
+	natsSub, err := js.jsCtx.PullSubscribe(pullSubject, durableId, subOpts...)
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			js.logg.Info("jetstream chain sub: shutdown signal received")
+			return nil
+		default:
+			events, err := natsSub.Fetch(1)
+			if err != nil {
+				if errors.Is(err, nats.ErrTimeout) {
+					// Supressed retry
+					js.logg.Error("jetstream chain sub: fetch NATS timeout", "error", err)
+					time.Sleep(backOffTimer)
+					continue
+				} else {
+					js.logg.Error("jetstream chain sub: fetch other error", "error", err)
+				}
+			}
+			if len(events) > 0 {
+				// TODO: Unmarshal
+				// TODO: UpdateOtxStatus
+			}
+		}
+
+	}
+}

internal/store/otx.go

@@ -57,3 +57,7 @@ func (s *PostgresStore) GetTxStatusByTrackingId(ctx context.Context, trackingId
 
 	return txs, nil
 }
+
+func (s *PostgresStore) UpdateOtxStatus(ctx context.Context, status string) error {
+	return nil
+}

internal/tasker/task/account_gift_gas.go

@@ -19,6 +19,7 @@ import (
 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
 		)
 

internal/tasker/task/account_gift_voucher.go

@@ -18,6 +18,7 @@ import (
 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
 		)
 

internal/tasker/task/account_refill_gas.go

@@ -21,8 +21,9 @@ import (
 func AccountRefillGasProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
 		var (
-			payload AccountPayload
 			balance big.Int
+			err     error
+			payload AccountPayload
 		)
 
 		if err := json.Unmarshal(t.Payload(), &payload); err != nil {

internal/tasker/task/account_register_onchain.go

@@ -19,6 +19,7 @@ import (
 func AccountRegisterOnChainProcessor(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
 		var (
+			err     error
 			payload AccountPayload
 		)
 

internal/tasker/task/sign_transfer.go

@@ -37,6 +37,7 @@ type (
 func SignTransfer(cu *custodial.Custodial) func(context.Context, *asynq.Task) error {
 	return func(ctx context.Context, t *asynq.Task) error {
 		var (
+			err     error
 			payload TransferPayload
 		)