✨ feat: implement payment streams and related commands

- Add a new `pay` command to the main application for executing policy-protected payments.
- Introduce `Stream` functionality in the `pkg/streaming` package to manage persistent payment channels.
- Create associated test cases for the `Stream` functionality.
- Update dependencies in `go.mod` and create `go.work` configuration to support new modules.

Author: nathfavour

cmd/settler/main.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/nathfavour/settlerengine/internal/domain"
 	"github.com/nathfavour/settlerengine/pkg/anyisland"
 	"github.com/nathfavour/settlerengine/pkg/crypto"
 	"github.com/nathfavour/settlerengine/pkg/storage"
@@ -38,6 +39,8 @@ func main() {
 		runProxy(os.Args[2:])
 	case "facilitator":
 		runFacilitator(os.Args[2:])
+	case "pay":
+		runPay(os.Args[2:])
 	case "help":
 		printUsage()
 	default:
@@ -54,9 +57,46 @@ func printUsage() {
 	fmt.Println("\nCommands:")
 	fmt.Println("  proxy        Start the x402 reverse proxy")
 	fmt.Println("  facilitator  Start the settlement facilitator daemon")
+	fmt.Println("  pay          Execute a policy-protected payment")
 	fmt.Println("  help         Show this help message")
 }
 
+func runPay(args []string) {
+	fs := flag.NewFlagSet("pay", flag.ExitOnError)
+	fs.String("rpc", "https://sepolia.base.org", "Ethereum RPC URL")
+	to := fs.String("to", "", "Recipient address")
+	amountStr := fs.String("amount", "0", "Amount in wei")
+	privKey := fs.String("key", "", "Private key (hex)")
+	maxPerTx := fs.String("max-per-tx", "1000000000000000000", "Max wei per transaction (1 ETH)")
+	fs.Parse(args)
+
+	if *to == "" || *privKey == "" {
+		log.Fatal("Recipient (-to) and Private Key (-key) are required")
+	}
+
+	chainID := big.NewInt(84532) // Base Sepolia
+	signer, err := crypto.NewSessionKeySigner(*privKey, chainID)
+	if err != nil {
+		log.Fatalf("Invalid signer: %v", err)
+	}
+
+	max, _ := new(big.Int).SetString(*maxPerTx, 10)
+	policy := domain.NewPaymentPolicy("cli-policy", max, nil, time.Time{})
+
+	policySigner := crypto.NewPolicySigner(signer, policy)
+	
+	amount, _ := new(big.Int).SetString(*amountStr, 10)
+	recipient := common.HexToAddress(*to)
+
+	// In a real CLI, we would use policySigner.GetTransactorWithPolicy
+	if err := policySigner.Check(amount, recipient); err != nil {
+		log.Fatalf("❌ Policy Blocked Payment: %v", err)
+	}
+
+	fmt.Printf("✅ Policy Approved Payment of %s wei to %s\n", amount.String(), recipient.Hex())
+	fmt.Println("Executing transaction... (Simulated for this demo)")
+}
+
 func runProxy(args []string) {
 	fs := flag.NewFlagSet("proxy", flag.ExitOnError)
 	target := fs.String("target", "http://localhost:8081", "Target URL to proxy to")

go.mod

@@ -5,7 +5,22 @@ go 1.24.0
 require (
 	github.com/ethereum/go-ethereum v1.16.8
 	github.com/google/uuid v1.6.0
+	github.com/gorilla/websocket v1.4.2
 	github.com/jackc/pgx/v5 v5.5.5
 	github.com/ncruces/go-sqlite3 v0.22.0
-	github.com/prometheus/client_golang v1.21.0
+	golang.org/x/crypto v0.36.0
+)
+
+require (
+	github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6 // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
+	github.com/holiman/uint256 v1.3.2 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/puddle/v2 v2.2.1 // indirect
+	github.com/ncruces/julianday v1.0.0 // indirect
+	github.com/tetratelabs/wazero v1.8.2 // indirect
+	golang.org/x/sync v0.17.0 // indirect
+	golang.org/x/sys v0.37.0 // indirect
+	golang.org/x/text v0.23.0 // indirect
 )

go.sum

@@ -0,0 +1,28 @@
+github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6 h1:1zYrtlhrZ6/b6SAjLSfKzWtdgqK0U+HtH/VcBWh1BaU=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/decred/dcrd/crypto/blake256 v1.0.0 h1:/8DMNYp9SGi5f0w7uCm6d6M4OU2rGFK09Y2A4Xv7EE0=
+github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 h1:YLtO71vCjJRCBcrPMtQ9nqBsqpA1m5sE92cU+pd5Mcc=
+github.com/ethereum/go-ethereum v1.16.8 h1:LLLfkZWijhR5m6yrAXbdlTeXoqontH+Ga2f9igY7law=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/holiman/uint256 v1.3.2 h1:a9EgMPSC1AAaj1SZL5zIQD3WbwTuHrMGOerLjGmM/TA=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
+github.com/jackc/pgx/v5 v5.5.5 h1:amBjrZVmksIdNjxGW/IiIMzxMKZFelXbUoPNb+8sjQw=
+github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
+github.com/ncruces/go-sqlite3 v0.22.0 h1:FkGSBhd0TY6e66k1LVhyEpA+RnG/8QkQNed5pjIk4cs=
+github.com/ncruces/julianday v1.0.0 h1:fH0OKwa7NWvniGQtxdJRxAgkBMolni2BjDHaWTxqt7M=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/tetratelabs/wazero v1.8.2 h1:yIgLR/b2bN31bjxwXHD8a3d+BogigR952csSDdLYEv4=
+golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
+golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
+golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
+golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

go.work

@@ -1 +1,9 @@
-// empty
+go 1.24.0
+
+use (
+	.
+	./apps/settler-proxy
+	./apps/settlerd
+	./core
+	./pkg
+)

go.work.sum

@@ -1 +1,7 @@
-// empty
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.5.5/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
+github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/ncruces/go-sqlite3 v0.22.0/go.mod h1:ueXOZXYZS2OFQirCU3mHneDwJm5fGKHrtccYBeGEV7M=
+github.com/ncruces/julianday v1.0.0/go.mod h1:Dusn2KvZrrovOMJuOt0TNXL6tB7U2E8kvza5fFc9G7g=
+github.com/tetratelabs/wazero v1.8.2/go.mod h1:yAI0XTsMBhREkM/YDAK/zNou3GoiAce1P6+rp/wQhjs=

pkg/crypto/signer.go

@@ -29,19 +29,24 @@ type SessionKeySigner struct {
 // PolicySigner wraps a SessionKeySigner with a policy.
 type PolicySigner struct {
 	*SessionKeySigner
-	policy Policy
+	Policy Policy
 }
 
 func NewPolicySigner(signer *SessionKeySigner, policy Policy) *PolicySigner {
 	return &PolicySigner{
 		SessionKeySigner: signer,
-		policy:           policy,
+		Policy:           policy,
 	}
 }
 
+// Check delegates the check to the underlying policy.
+func (s *PolicySigner) Check(amount *big.Int, recipient common.Address) error {
+	return s.Policy.Check(amount, recipient)
+}
+
 // GetTransactorWithPolicy returns a transactor only if the policy allows the transaction.
 func (s *PolicySigner) GetTransactorWithPolicy(ctx context.Context, client *ethclient.Client, amount *big.Int, recipient common.Address) (*bind.TransactOpts, error) {
-	if err := s.policy.Check(amount, recipient); err != nil {
+	if err := s.Policy.Check(amount, recipient); err != nil {
 		return nil, fmt.Errorf("policy violation: %w", err)
 	}
 
@@ -55,7 +60,7 @@ func (s *PolicySigner) GetTransactorWithPolicy(ctx context.Context, client *ethc
 	// Wrap the signer to record the update on success
 	// Note: In a real implementation, we'd only record once the tx is confirmed.
 	// For this lightweight fix, we'll record it when the transactor is requested.
-	s.policy.RecordUpdate(amount)
+	s.Policy.RecordUpdate(amount)
 
 	return auth, nil
 }

pkg/streaming/stream.go

@@ -0,0 +1,105 @@
+package streaming
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/nathfavour/settlerengine/pkg/crypto"
+)
+
+// Stream represents a persistent payment channel between an agent and a service.
+type Stream struct {
+	ID        string
+	Recipient common.Address
+	Asset     common.Address
+	Rate      *big.Int      // Amount per interval
+	Interval  time.Duration
+	
+	mu        sync.RWMutex
+	LastPulse time.Time
+	TotalPaid *big.Int
+	Active    bool
+}
+
+// NewStream creates a new payment stream.
+func NewStream(id string, recipient, asset common.Address, rate *big.Int, interval time.Duration) *Stream {
+	return &Stream{
+		ID:        id,
+		Recipient: recipient,
+		Asset:     asset,
+		Rate:      rate,
+		Interval:  interval,
+		TotalPaid: big.NewInt(0),
+		Active:    true,
+	}
+}
+
+// ValidatePulse checks if a pulse signature is valid and covers the required interval.
+func (s *Stream) ValidatePulse(intent crypto.IntentToPay, signature string, domain crypto.DomainParams) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if !s.Active {
+		return fmt.Errorf("stream is inactive")
+	}
+
+	// 1. Verify Signature
+	_, err := crypto.VerifyIntentToPay(intent, signature, domain)
+	if err != nil {
+		return fmt.Errorf("invalid pulse signature: %w", err)
+	}
+
+	// 2. Verify Recipient and Asset
+	if intent.Recipient != s.Recipient.Hex() {
+		return fmt.Errorf("wrong recipient in pulse")
+	}
+	if intent.Asset != s.Asset.Hex() {
+		return fmt.Errorf("wrong asset in pulse")
+	}
+
+	// 3. Verify Amount (must be >= Rate)
+	amount, ok := new(big.Int).SetString(intent.Amount, 10)
+	if !ok || amount.Cmp(s.Rate) < 0 {
+		return fmt.Errorf("insufficient amount in pulse")
+	}
+
+	// 4. Update state
+	s.LastPulse = time.Now()
+	s.TotalPaid = new(big.Int).Add(s.TotalPaid, amount)
+
+	return nil
+}
+
+// Monitor monitors the stream and deactivates it if a pulse is missed.
+func (s *Stream) Monitor(ctx context.Context, gracePeriod time.Duration) {
+	ticker := time.NewTicker(s.Interval / 2)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.mu.RLock()
+			lastPulse := s.LastPulse
+			active := s.Active
+			s.mu.RUnlock()
+
+			if !active {
+				return
+			}
+
+			if !lastPulse.IsZero() && time.Since(lastPulse) > (s.Interval + gracePeriod) {
+				s.mu.Lock()
+				s.Active = false
+				s.mu.Unlock()
+				fmt.Printf("🔴 Stream %s deactivated: pulse timeout\n", s.ID)
+				return
+			}
+		}
+	}
+}

pkg/streaming/stream_test.go

@@ -0,0 +1,43 @@
+package streaming
+
+import (
+	"context"
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+)
+
+func TestStream(t *testing.T) {
+	recipient := common.HexToAddress("0x1234567890123456789012345678901234567890")
+	asset := common.HexToAddress("0x036CbD53842c5426634e7929541eC2318f3dCF7e")
+	rate := big.NewInt(1000)
+	interval := 100 * time.Millisecond
+
+	s := NewStream("test-stream", recipient, asset, rate, interval)
+
+	t.Run("InitialState", func(t *testing.T) {
+		if !s.Active {
+			t.Error("expected stream to be active")
+		}
+	})
+
+	t.Run("MonitorTimeout", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+		defer cancel()
+
+		s.LastPulse = time.Now()
+		go s.Monitor(ctx, 50*time.Millisecond)
+
+		time.Sleep(300 * time.Millisecond)
+
+		s.mu.RLock()
+		active := s.Active
+		s.mu.RUnlock()
+
+		if active {
+			t.Error("expected stream to be deactivated after timeout")
+		}
+	})
+}