🚧 fees v2

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/verifier/plugin/keysign"
 	"github.com/vultisig/verifier/plugin/tasks"
@@ -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: ${ETH_PROVIDER_URL}
+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)
+}

plugin/fees/FEES.md

@@ -0,0 +1,162 @@
+# Fees plugin
+
+The fees pluigin is responsible for collecting user based fees incurred from other plugins
+
+## Overview
+
+As this is a 2 part system, and the fees plugin works differently from other plugins. Here is an overview of the current architecture.
+
+### Verifier
+
+The verifier is responsible for tracking the fees which need to be collected. This is achieved by creating a `billing` record against each plugin policy. The `billing` record contains information which is encoded in the `FeePolicy` struct of the recipe that a user signs. There may be zero, one, or several fee policies, and hence `billing` records connected to a `plugin policy`.
+
+When a new plugin policy is created this data is extracted, verified (against the `pricings` (definitions) table) and then inserted accordingly in the db. It is also synced to the plugin server and if it fails everything fails. The following types of `billing` record exist. They are all recorded as enums in the db and in the implementation
+
+- `once` - **fully implemented** one billing record and one `fee` record are created on policy insert.
+- `recurring` - **mostly implemented** the db views and code is there but there is no scheduled job to create the fee records at intervals.
+- `tx` - **barely implemented** the enum types exist but the code to incur fees upon certain tx based transactions is not yet there.
+
+#### Fee records
+
+Each fee has a unique ID in the system and **must** be connected to a `billing` record. Several fees may share the same `billing` record in the cases of recurrent or tx based billing records. In the case of one time fees, the fee is incurred instantly on successful install, otherwise the fee creations will be handled via trigger. Fees are known to be collected successfully by the use of a `collected_at` attribute and the `charged_at` as being set as the date the fee was incurred which is pre-empted to be needed for audit purposes.
+
+#### Tx signing
+
+- *currently pending pr* - fee ids are passed along with a signing request which are used to verify that the amount being requested for a signature is the same.
+- *current* - the system checks for all fees which are due to be collected and if the amount doesn't match it rejects a sign request.
+
+The signing request from the plugin server to the verifier is parsed and put through the above checks if the `policy_id` is that of a fee plugin.
+
+### Plugin
+
+In the `v2` implementation there are 3 distinct processes for handling fees. 
+
+`loading` - pulls in the fees from the verifier and assigns them to a `fee run` (more later)
+`transaction` - iterates through each of the fee runs, builds transactions, sends them to the verifier to be signed and then broadcasts them
+`post` - checks the sent transactions and if successful updates the verifier which marks a value as `collected_at` **pending pr** (at the same time a `treasury_ledger` table is appended to with the fees which go to a developer and vultisig.)
+
+#### Fee Runs
+
+A `fee run` is a logical grouping of fees. They have various states depending on their lifecycle. When loading fees if a fee is found which has not yet been transacted *and* there is a `pending` (aka draft) fee run then the fee is included with it. If no fee run is detected then a new fee run is created.
+
+The db structure are different between verifier and plugin here due to a separation of concerns. Verifier needs to create fee entries, track their state across their lifecycle and handle treasury output from them. The plugin server simply needs to group them and track their ids and amounts.
+
+### Detailed Workflow
+
+#### 1. Fee Loading Process (`LoadFees`)
+
+**Trigger**: Scheduled cron job (configurable interval, default: every 10 minutes)
+
+**Process**:
+1. Retrieves all fee policies from database
+2. For each policy, queries verifier API for pending fees
+3. Validates fee amounts and consistency
+4. Creates or updates fee runs in `draft` status
+5. Adds individual fees to fee runs
+6. Uses semaphore for concurrent processing (configurable limit).
+7. Will only run when one of the other 3 processes isn't running
+
+**Key Features**:
+- Concurrent processing with semaphore limiting
+- Duplicate fee detection and prevention
+- Automatic fee run creation and management
+- Error handling and rollback capabilities
+
+#### 2. Transaction Handling (`HandleTransactions`)
+
+**Trigger**: Scheduled cron job (configurable interval, default: fridays weekly)
+
+**Process**:
+1. Retrieves all fee runs in `draft` status
+2. For each valid fee run:
+   - Generates Ethereum ERC20 USDC transfer transaction
+   - Creates keysign request with transaction data
+   - Initiates signing process through TSS
+   - Broadcasts signed transaction to blockchain
+   - Updates fee run status to `sent`
+
+**Transaction Details**:
+- **Token**: USDC (ERC20)
+- **Chain**: Ethereum (chainId: 1)
+- **Recipient**: Vultisig Treasury (resolved via magic constants)
+- **Gas Limit**: 65,000 (typical ERC20 transfer upper bound)
+
+#### 3. Post-Transaction Processing (`HandlePostTx`)
+
+**Trigger**: Scheduled cron job (configurable interval, default: every 10 minutes)
+
+**Process**:
+1. Monitors all fee runs in `sent` status
+2. Checks transaction receipts on blockchain
+3. Validates confirmation count against configured threshold
+4. Updates verifier with collection status
+5. Marks fee runs as `completed` or `failed`
+
+**Confirmation Logic**:
+- Waits for configurable number of confirmations
+- Handles transaction failures and rebroadcast scenarios
+- Maintains state consistency between plugin and verifier
+
+### Job Scheduling & Configuration
+
+#### Asynq Task Types
+
+1. **`fees:load`** - Fee loading from verifier
+2. **`fees:transaction`** - Transaction creation and broadcasting  
+3. **`fees:post_tx`** - Post-transaction status checking
+
+#### Policy Validation
+
+The system validates fee policies against strict criteria:
+
+1. **Resource Validation**: Only `ethereum.erc20.transfer` operations allowed
+2. **Recipient Validation**: Must use `VULTISIG_TREASURY` magic constant
+3. **Amount Constraints**: Maximum fee amount enforcement
+4. **Recipe Schema**: Validates against predefined recipe specification
+
+#### Transaction Security
+
+1. **Mutex Protection**: Prevents concurrent transaction operations
+2. **Amount Verification**: Cross-validates amounts between verifier and plugin
+3. **Signature Validation**: Uses TSS for secure transaction signing
+4. **Rollback Mechanisms**: Database transactions ensure consistency
+
+### API Integration
+
+#### Verifier API Endpoints
+
+1. **`GetPublicKeysFees(publicKey)`** - Retrieves pending fees for a vault
+2. **`MarkFeeAsCollected(txHash, timestamp, feeIds...)`** - Marks fees as collected
+
+### Monitoring & Observability
+
+#### Logging Structure
+
+All operations use structured logging with consistent field names:
+- `publicKey`: Vault public key
+- `feeId`/`feeIds`: Individual fee identifiers
+- `runId`: Fee run identifier  
+- `policyId`: Plugin policy identifier
+- `tx_hash`: Blockchain transaction hash
+
+### Implementation Status
+
+#### Fully Implemented Features
+- `once` billing type - Single fee collection on policy creation
+- Fee loading and aggregation system
+- ERC20 USDC transaction generation
+- TSS-based transaction signing
+- Post-transaction confirmation tracking
+- Database schema and migrations
+
+#### Partially Implemented Features  
+- `recurring` billing type - Database structure exists, scheduling incomplete
+- Transaction rebroadcast logic - Framework present, full implementation pending
+- Failed transaction recovery - Basic structure, comprehensive handling needed
+
+#### Future Enhancements
+- `tx` billing type - Transaction-based fee collection
+- Multi-token support beyond USDC
+- Cross-chain fee collection capabilities
+- Advanced retry and recovery mechanisms
+