Support advertised chain options in the web wallet flow 228

gateway/cache.go

@@ -164,6 +164,24 @@ func CacheMiddleware() gin.HandlerFunc {
 				c.Abort()
 				return
 			}
+
+			// === Fix for #241: Prevent replay attacks on cache hits ===
+			used, err := markTransactionUsed(c.Request.Context(), signature)
+			if err != nil {
+				respondError(c, 500, "internal_error", fmt.Errorf("failed to check transaction replay: %w", err))
+				c.Abort()
+				return
+			}
+			if used {
+				respondVerificationFailure(c, &VerifyResponse{
+					ErrorCode: "nonce_already_used",
+					Error:     "Transaction already used",
+				})
+				c.Abort()
+				return
+			}
+			// ==========================================================
+
 			verificationTotal.WithLabelValues("success").Inc()
 			// Payment Verified. Store verification for downstream if needed (though we abort)
 			c.Set("payment_verification", verifyResp)
@@ -174,9 +192,9 @@ func CacheMiddleware() gin.HandlerFunc {
 			// Generate receipt for cache hit using current request and cached result.
 			// Note: request_hash matches current request, response is from cache,
 			// but both are cryptographically valid since cache key ensures identical text.
-			if err := generateAndSendReceipt(c, *paymentCtx, verifyResp.RecoveredAddress, requestBody, cached.Result); err != nil {
+			if err := streamResultAndReceipt(c, *paymentCtx, verifyResp.RecoveredAddress, requestBody, cached.Result); err != nil {
 				log.Printf("Failed to send cached response receipt: %v", err)
-				// generateAndSendReceipt already sent an error response (500)
+				// streamResultAndReceipt already sent an error response
 			}
 			c.Abort()
 			return
@@ -204,14 +222,13 @@ func CacheMiddleware() gin.HandlerFunc {
 		// Handler finished. Check status and extract result with proper locking
 		writer.mu.RLock()
 		statusCode := writer.ResponseWriter.Status()
-		bodyBytes := writer.body.Bytes()
 		writer.mu.RUnlock()
 
 		if statusCode == 200 {
-			// Response format: {"result": "...", "receipt": ...}
-			var resp map[string]interface{}
-			if err := json.Unmarshal(bodyBytes, &resp); err == nil {
-				if result, ok := resp["result"].(string); ok {
+			// Instead of parsing the SSE stream from bodyBytes, we retrieve the full summary
+			// which handleSummarize saves in the gin context.
+			if fullSummaryVal, exists := c.Get("full_summary"); exists {
+				if result, ok := fullSummaryVal.(string); ok {
 					// Store asynchronously with a deadline to prevent indefinite goroutines
 					go func(k, v string) {
 						ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

gateway/cache_integration_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -198,18 +199,33 @@ func TestCacheIntegration_FullFlow(t *testing.T) {
 	}
 
 	// Verify Body
-	var resp1, resp2 map[string]interface{}
-	if err := json.Unmarshal(w1.Body.Bytes(), &resp1); err != nil {
-		t.Fatalf("Failed to unmarshal response 1: %v", err)
-	}
-	if err := json.Unmarshal(w2.Body.Bytes(), &resp2); err != nil {
-		t.Fatalf("Failed to unmarshal response 2: %v", err)
+	parseSSEResultForTest := func(body []byte) string {
+		lines := strings.Split(string(body), "\n")
+		var result string
+		for _, line := range lines {
+			if strings.HasPrefix(line, "data: ") {
+				data := strings.TrimSpace(strings.TrimPrefix(line, "data: "))
+				if data == "[DONE]" || data == "" {
+					continue
+				}
+				var chunk map[string]interface{}
+				if err := json.Unmarshal([]byte(data), &chunk); err == nil {
+					if text, ok := chunk["text"].(string); ok {
+						result += text
+					}
+				}
+			}
+		}
+		return result
 	}
 
-	if val, ok := resp1["result"].(string); !ok || val != "AI Summary Result" {
-		t.Errorf("Unexpected result 1: %v", resp1["result"])
+	result1 := parseSSEResultForTest(w1.Body.Bytes())
+	result2 := parseSSEResultForTest(w2.Body.Bytes())
+
+	if result1 != "AI Summary Result" {
+		t.Errorf("Unexpected result 1: %v", result1)
 	}
-	if val, ok := resp2["result"].(string); !ok || val != "AI Summary Result" {
-		t.Errorf("Unexpected result 2: %v", resp2["result"])
+	if result2 != "AI Summary Result" {
+		t.Errorf("Unexpected result 2: %v", result2)
 	}
 }

gateway/errors_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"gateway/internal/ai"
 	"github.com/alicebob/miniredis/v2"
 	"github.com/gin-gonic/gin"
 	"github.com/redis/go-redis/v9"
@@ -24,6 +25,10 @@ func (p failingProvider) Generate(context.Context, string) (string, error) {
 	return "", p.err
 }
 
+func (p failingProvider) GenerateStream(context.Context, string) (ai.Stream, error) {
+	return nil, p.err
+}
+
 func newSummarizeTestRouter() *gin.Engine {
 	gin.SetMode(gin.TestMode)
 	r := gin.New()

gateway/internal/ai/ollama.go

@@ -77,3 +77,58 @@ func (p *OllamaProvider) Generate(ctx context.Context, text string) (string, err
 
 	return response, nil
 }
+
+type ollamaStream struct {
+	resp    *http.Response
+	decoder *json.Decoder
+}
+
+func (s *ollamaStream) Recv() (string, error) {
+	var chunk map[string]interface{}
+	if err := s.decoder.Decode(&chunk); err != nil {
+		return "", err // will return io.EOF when stream ends
+	}
+	if response, ok := chunk["response"].(string); ok {
+		return response, nil
+	}
+	return "", fmt.Errorf("invalid response from Ollama: missing response field")
+}
+
+func (s *ollamaStream) Close() error {
+	return s.resp.Body.Close()
+}
+
+func (p *OllamaProvider) GenerateStream(ctx context.Context, text string) (Stream, error) {
+	prompt := fmt.Sprintf("Summarize this text in 2 sentences: %s", text)
+
+	reqBody, _ := json.Marshal(map[string]interface{}{
+		"model":  p.model,
+		"prompt": prompt,
+		"stream": true,
+	})
+
+	req, err := http.NewRequestWithContext(ctx, "POST", p.url+"/api/generate", bytes.NewBuffer(reqBody))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Ollama request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		if errors.Is(err, context.DeadlineExceeded) || ctx.Err() == context.DeadlineExceeded {
+			return nil, context.DeadlineExceeded
+		}
+		return nil, fmt.Errorf("failed to connect to Ollama: %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		return nil, fmt.Errorf("ollama returned status %d: %s", resp.StatusCode, string(body))
+	}
+
+	return &ollamaStream{
+		resp:    resp,
+		decoder: json.NewDecoder(resp.Body),
+	}, nil
+}

gateway/internal/ai/openrouter.go

@@ -1,6 +1,7 @@
 package ai
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -100,3 +101,88 @@ func (p *OpenRouterProvider) Generate(ctx context.Context, text string) (string,
 
 	return content, nil
 }
+
+type openRouterStream struct {
+	resp   *http.Response
+	reader *bufio.Reader
+}
+
+func (s *openRouterStream) Recv() (string, error) {
+	for {
+		line, err := s.reader.ReadBytes('\n')
+		if err != nil {
+			return "", err
+		}
+		line = bytes.TrimSpace(line)
+		if len(line) == 0 {
+			continue
+		}
+		if bytes.HasPrefix(line, []byte("data: ")) {
+			data := bytes.TrimPrefix(line, []byte("data: "))
+			if string(data) == "[DONE]" {
+				return "", io.EOF
+			}
+			var chunk map[string]interface{}
+			if err := json.Unmarshal(data, &chunk); err != nil {
+				continue // skip invalid json
+			}
+			choices, ok := chunk["choices"].([]interface{})
+			if !ok || len(choices) == 0 {
+				continue
+			}
+			choice, ok := choices[0].(map[string]interface{})
+			if !ok {
+				continue
+			}
+			delta, ok := choice["delta"].(map[string]interface{})
+			if !ok {
+				continue
+			}
+			if content, ok := delta["content"].(string); ok {
+				return content, nil
+			}
+		}
+	}
+}
+
+func (s *openRouterStream) Close() error {
+	return s.resp.Body.Close()
+}
+
+func (p *OpenRouterProvider) GenerateStream(ctx context.Context, text string) (Stream, error) {
+	prompt := fmt.Sprintf("Summarize this text in 2 sentences: %s", text)
+
+	reqBody, _ := json.Marshal(map[string]interface{}{
+		"model": p.model,
+		"messages": []map[string]string{
+			{"role": "user", "content": prompt},
+		},
+		"stream": true,
+	})
+
+	req, err := http.NewRequestWithContext(ctx, "POST", p.url, bytes.NewBuffer(reqBody))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create OpenRouter request: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+p.apiKey)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		if errors.Is(err, context.DeadlineExceeded) || ctx.Err() == context.DeadlineExceeded {
+			return nil, context.DeadlineExceeded
+		}
+		return nil, err
+	}
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		body, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		return nil, fmt.Errorf("openrouter returned status %d: %s", resp.StatusCode, string(body))
+	}
+
+	return &openRouterStream{
+		resp:   resp,
+		reader: bufio.NewReader(resp.Body),
+	}, nil
+}

gateway/internal/ai/provider.go

@@ -6,10 +6,20 @@ import (
 	"os"
 )
 
+// Stream defines the interface for streaming AI responses
+type Stream interface {
+	// Recv returns the next token or an error (io.EOF when done)
+	Recv() (string, error)
+	// Close releases any resources associated with the stream
+	Close() error
+}
+
 // Provider defines the interface for AI service providers
 type Provider interface {
 	// Generate takes a context and prompt text, returns the AI-generated response
 	Generate(ctx context.Context, prompt string) (string, error)
+	// GenerateStream takes a context and prompt text, returns a Stream of generated tokens
+	GenerateStream(ctx context.Context, prompt string) (Stream, error)
 }
 
 // NewProvider creates an AI provider based on the AI_PROVIDER environment variable