Fees logic separation

cmd/fees/worker/main.go

@@ -2,11 +2,13 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/hibiken/asynq"
+	"github.com/robfig/cron/v3"
 	"github.com/sirupsen/logrus"
 	"github.com/vultisig/plugin/internal/keysign"
 	"github.com/vultisig/verifier/tx_indexer"
@@ -20,6 +22,45 @@ import (
 	"github.com/vultisig/plugin/storage/postgres"
 )
 
+func startLoadingFees(asynqClient *asynq.Client, logger *logrus.Logger) {
+	logger.Info("Loading fees")
+	payload, err := json.Marshal(fees.FeeCollectionFormat{
+		FeeCollectionType: fees.FeeCollectionTypeAll,
+	})
+	if err != nil {
+		logger.WithError(err).Error("Failed to marshal fee loading config in demo run")
+		return
+	}
+	asynqClient.Enqueue(
+		asynq.NewTask(fees.TypeFeeLoad, payload),
+		asynq.MaxRetry(0),
+		asynq.Timeout(2*time.Minute),
+		asynq.Retention(5*time.Minute),
+		asynq.Queue(tasks.QUEUE_NAME))
+}
+
+func startTransactingFees(asynqClient *asynq.Client, logger *logrus.Logger) {
+	logger.Info("Transacting fees")
+	payload := make([]byte, 0)
+	asynqClient.Enqueue(
+		asynq.NewTask(fees.TypeFeeTransact, payload),
+		asynq.MaxRetry(0),
+		asynq.Timeout(2*time.Minute),
+		asynq.Retention(5*time.Minute),
+		asynq.Queue(tasks.QUEUE_NAME))
+}
+
+func startPostTx(asynqClient *asynq.Client, logger *logrus.Logger) {
+	logger.Info("Checking status")
+	payload := make([]byte, 0)
+	asynqClient.Enqueue(
+		asynq.NewTask(fees.TypeFeePostTx, payload),
+		asynq.MaxRetry(0),
+		asynq.Timeout(2*time.Minute),
+		asynq.Retention(5*time.Minute),
+		asynq.Queue(tasks.QUEUE_NAME))
+}
+
 func main() {
 	ctx := context.Background()
 
@@ -132,9 +173,33 @@ func main() {
 	mux.HandleFunc(tasks.TypeReshareDKLS, vaultService.HandleReshareDKLS)
 
 	//Plugin specific functions.
-	mux.HandleFunc(fees.TypeFeeCollection, feePlugin.HandleCollections)
+	mux.HandleFunc(fees.TypeFeeLoad, feePlugin.LoadFees)
+	mux.HandleFunc(fees.TypeFeeTransact, feePlugin.HandleTransactions)
+	mux.HandleFunc(fees.TypeFeePostTx, feePlugin.HandlePostTx)
+
+	// Load fees every 10 minutes
+	loadFees := cron.New()
+	loadFees.AddFunc(feePluginConfig.Jobs.Load.Cronexpr, func() {
+		startLoadingFees(asynqClient, logger)
+	})
+	loadFees.Start()
+
+	// Transact fees every Friday at 12:00 PM
+	transactFees := cron.New()
+	transactFees.AddFunc(feePluginConfig.Jobs.Transact.Cronexpr, func() {
+		startTransactingFees(asynqClient, logger)
+	})
+	transactFees.Start()
+
+	// // Update verifier every 10 minutes
+	updateVerifier := cron.New()
+	updateVerifier.AddFunc(feePluginConfig.Jobs.Post.Cronexpr, func() {
+		startPostTx(asynqClient, logger)
+	})
+	updateVerifier.Start()
 
 	logger.Info("Starting asynq listener")
+
 	if err := srv.Run(mux); err != nil {
 		panic(fmt.Errorf("could not run server: %w", err))
 	}

etc/vultisig/fee.yml

@@ -1,9 +1,16 @@
 type: fee
 version: 1.0.0
-rpc_url: https://ethereum.publicnode.com/
-collector_whitelist_addresses:
-    - 0x7d760c17d798a7A9a4c4AcAf311A02dC95972503
-    - 0x5BB06B9C5e4f7624Df7100Badb2F5AA1C86d8498
-collector_address: 0x7d760c17d798a7A9a4c4AcAf311A02dC95972503
 usdc_address: 0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48
 verifier_token: localhost-fee-apikey
+eth_provider: https://eth-mainnet.g.alchemy.com/v2/HAtIwB5y82TNVwaHkwcMj
+chain_id: 1
+jobs:
+    load:
+        cronexpr: "@every 2m"
+        max_concurrent_jobs: 10
+    transact:
+        cronexpr: "0 12 * * 5"
+        max_concurrent_jobs: 10
+    post:
+        cronexpr: "@every 5m"
+        max_concurrent_jobs: 10

internal/types/fees.go

@@ -11,8 +11,10 @@ import (
 type FeeRunState string
 
 const (
-	FeeRunStateDraft FeeRunState = "draft"
-	FeeRunStateSent  FeeRunState = "sent"
+	FeeRunStateDraft   FeeRunState = "draft"
+	FeeRunStateSent    FeeRunState = "sent"
+	FeeRunStateSuccess FeeRunState = "completed"
+	FeeRunStateFailed  FeeRunState = "failed"
 )
 
 // individual fee record in the db
@@ -29,8 +31,9 @@ type FeeRun struct {
 	Status      FeeRunState `db:"status"`
 	CreatedAt   time.Time   `db:"created_at"`
 	UpdatedAt   time.Time   `db:"updated_at"`
-	TxID        *uuid.UUID  `db:"tx_id"`
+	TxHash      *string     `db:"tx_hash"`
 	PolicyID    uuid.UUID   `db:"policy_id"`
 	TotalAmount int         `db:"total_amount"`
 	FeeCount    int         `db:"fee_count"`
+	Fees        []Fee       `db:"fees"`
 }

internal/verifierapi/fees.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"time"
 
 	"github.com/google/uuid"
 )
@@ -55,3 +56,29 @@ func (v *VerifierApi) GetPublicKeysFees(ecdsaPublicKey string) (*FeeHistoryDto,
 
 	return &feeHistory.Data, nil
 }
+
+func (v *VerifierApi) MarkFeeAsCollected(txHash string, collectedAt time.Time, feeIds ...uuid.UUID) error {
+
+	var body = struct {
+		IDs         []uuid.UUID `json:"ids"`
+		TxHash      string      `json:"tx_hash"`
+		CollectedAt time.Time   `json:"collected_at"`
+	}{
+		IDs:         feeIds,
+		TxHash:      txHash,
+		CollectedAt: collectedAt,
+	}
+
+	url := "/fees/collected"
+	response, err := v.postAuth(url, body)
+	if err != nil {
+		return fmt.Errorf("failed to mark fee as collected: %w", err)
+	}
+	defer response.Body.Close()
+
+	if response.StatusCode != http.StatusOK {
+		return fmt.Errorf("failed to mark fee as collected, status code: %d", response.StatusCode)
+	}
+
+	return nil
+}

internal/verifierapi/verifierapi.go

@@ -1,6 +1,8 @@
 package verifierapi
 
 import (
+	"bytes"
+	"encoding/json"
 	"net/http"
 
 	"github.com/sirupsen/logrus"
@@ -49,3 +51,20 @@ func (v *VerifierApi) getAuth(endpoint string) (*http.Response, error) {
 	r.Header.Set("Authorization", "Bearer "+v.token)
 	return v.client.Do(r)
 }
+
+func (v *VerifierApi) postAuth(endpoint string, body any) (*http.Response, error) {
+	jsonBody, err := json.Marshal(body)
+	if err != nil {
+		return nil, err
+	}
+
+	request, err := http.NewRequest(http.MethodPost, v.url+endpoint, bytes.NewBuffer(jsonBody))
+	if err != nil {
+		return nil, err
+	}
+
+	request.Header.Set("Content-Type", "application/json")
+	request.Header.Set("Authorization", "Bearer "+v.token)
+
+	return v.client.Do(request)
+}