fix: harden terminal-scoped analytics contracts

Keep JSON and bounded CSV exports on one canonical predicate, replace replayable offsets with encrypted tenant/filter-bound keyset cursors, and suppress merchant-grain peers under terminal scope. Validate period filters and align OpenAPI and capability specs with runtime behavior.
2026-08-20 21:29:18 +03:30
parent 857f29635e
commit d6de48fd15
14 changed files with 577 additions and 103 deletions
--- a/internal/analytics/cursor.go
+++ b/internal/analytics/cursor.go
@@ -0,0 +1,108 @@
+package analytics
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"io"
+	"time"
+)
+
+var ErrInvalidCursor = errors.New("analytics: invalid cursor")
+
+type detailCursorPosition struct {
+	CreatedAt time.Time
+	SessionID int64
+}
+
+type detailCursorPayload struct {
+	Version   int       `json:"v"`
+	Binding   string    `json:"b"`
+	CreatedAt time.Time `json:"createdAt"`
+	SessionID int64     `json:"sessionId"`
+	ExpiresAt time.Time `json:"expiresAt"`
+}
+
+type detailCursorBinding struct {
+	MerchantID   int64  `json:"merchantId"`
+	PeriodID     string `json:"periodId"`
+	View         string `json:"view"`
+	FailureStage string `json:"failureStage"`
+	TerminalID   string `json:"terminalId"`
+	Status       string `json:"status"`
+	Minimum      *int64 `json:"minimum"`
+	Maximum      *int64 `json:"maximum"`
+}
+
+func (s *Service) encodeDetailCursor(merchantID int64, period Period, filter DetailFilter, createdAt time.Time, sessionID int64) (string, error) {
+	payload := detailCursorPayload{
+		Version:   1,
+		Binding:   detailCursorBindingDigest(merchantID, period, filter),
+		CreatedAt: createdAt.UTC(),
+		SessionID: sessionID,
+		ExpiresAt: time.Now().UTC().Add(24 * time.Hour),
+	}
+	encoded, err := json.Marshal(payload)
+	if err != nil {
+		return "", err
+	}
+	aead, err := s.cursorAEAD()
+	if err != nil {
+		return "", err
+	}
+	nonce := make([]byte, aead.NonceSize())
+	if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
+		return "", err
+	}
+	sealed := aead.Seal(nonce, nonce, encoded, nil)
+	return base64.RawURLEncoding.EncodeToString(sealed), nil
+}
+
+func (s *Service) decodeDetailCursor(merchantID int64, period Period, filter DetailFilter, cursor string) (detailCursorPosition, error) {
+	sealed, err := base64.RawURLEncoding.DecodeString(cursor)
+	if err != nil {
+		return detailCursorPosition{}, ErrInvalidCursor
+	}
+	aead, err := s.cursorAEAD()
+	if err != nil || len(sealed) < aead.NonceSize() {
+		return detailCursorPosition{}, ErrInvalidCursor
+	}
+	nonce, ciphertext := sealed[:aead.NonceSize()], sealed[aead.NonceSize():]
+	payloadBytes, err := aead.Open(nil, nonce, ciphertext, nil)
+	if err != nil {
+		return detailCursorPosition{}, ErrInvalidCursor
+	}
+	var payload detailCursorPayload
+	if err := json.Unmarshal(payloadBytes, &payload); err != nil || payload.Version != 1 || payload.SessionID <= 0 || payload.CreatedAt.IsZero() || time.Now().UTC().After(payload.ExpiresAt) {
+		return detailCursorPosition{}, ErrInvalidCursor
+	}
+	wantBinding := detailCursorBindingDigest(merchantID, period, filter)
+	if payload.Binding != wantBinding {
+		return detailCursorPosition{}, ErrInvalidCursor
+	}
+	return detailCursorPosition{CreatedAt: payload.CreatedAt.UTC(), SessionID: payload.SessionID}, nil
+}
+
+func (s *Service) cursorAEAD() (cipher.AEAD, error) {
+	key := sha256.Sum256(s.cursorSecret)
+	block, err := aes.NewCipher(key[:])
+	if err != nil {
+		return nil, err
+	}
+	return cipher.NewGCM(block)
+}
+
+func detailCursorBindingDigest(merchantID int64, period Period, filter DetailFilter) string {
+	binding := detailCursorBinding{
+		MerchantID: merchantID, PeriodID: period.PeriodID, View: filter.View,
+		FailureStage: filter.FailureStage, TerminalID: filter.TerminalID, Status: filter.Status,
+		Minimum: filter.MinAmountRial, Maximum: filter.MaxAmountRial,
+	}
+	encoded, _ := json.Marshal(binding)
+	digest := sha256.Sum256(encoded)
+	return base64.RawURLEncoding.EncodeToString(digest[:])
+}
--- a/internal/analytics/details.go
+++ b/internal/analytics/details.go
@@ -2,16 +2,33 @@ package analytics

 import (
 	"context"
-	"encoding/base64"
 	"errors"
 	"fmt"
-	"strconv"
 	"strings"
 	"time"

 	"github.com/jackc/pgx/v5"
 )

+const MaxExportRows = 100_000
+
+var ErrExportTooLarge = errors.New("analytics: export exceeds row limit")
+
+const detailFilterPredicate = `
+	ps.merchant_id=$1 AND ps.created_at>=$2 AND ps.created_at<$3
+	AND ($4='ALL'
+		OR ($4='UNVERIFIED_PAID' AND ps.status='Paid')
+		OR ($4='FAILED' AND ps.status='Failed')
+		OR ($4='RETRIED' AND (SELECT count(*) FROM payment_attempts retry_attempts WHERE retry_attempts.session_id=ps.id)>1))
+	AND ($5='' OR t.source_key=$5)
+	AND ($6='' OR
+		($6='NO_ATTEMPT' AND NOT EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id)) OR
+		($6='REACHED_BANK' AND EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id AND failure_attempts.status='InBank')) OR
+		($6='OTHER_OBSERVED' AND EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id) AND NOT EXISTS (SELECT 1 FROM payment_attempts failure_attempts WHERE failure_attempts.session_id=ps.id AND failure_attempts.status='InBank')))
+	AND ($7='' OR ps.status=$7)
+	AND ($8::bigint IS NULL OR ps.amount_rial >= $8)
+	AND ($9::bigint IS NULL OR ps.amount_rial <= $9)`
+
 type DetailFilter struct {
 	View          string
 	FailureStage  string
@@ -70,6 +87,12 @@ type Detail struct {
 	Attempts      []Attempt  `json:"attempts"`
 }

+type ExportDetail struct {
+	Detail
+	LastAttemptStatus *string
+	LastAttemptAt     *time.Time
+}
+
 func (s *Service) Details(ctx context.Context, merchantID int64, period Period, filter DetailFilter) ([]Detail, string, error) {
 	if filter.Limit <= 0 {
 		filter.Limit = 25
@@ -77,39 +100,22 @@ func (s *Service) Details(ctx context.Context, merchantID int64, period Period,
 	if err := filter.Validate(); err != nil {
 		return nil, "", err
 	}
-	offset := 0
+	var cursorCreatedAt *time.Time
+	var cursorSessionID int64
 	if filter.Cursor != "" {
-		decoded, err := base64.RawURLEncoding.DecodeString(filter.Cursor)
+		position, err := s.decodeDetailCursor(merchantID, period, filter, filter.Cursor)
 		if err != nil {
-			return nil, "", errors.New("analytics: invalid cursor")
+			return nil, "", err
 		}
-		offset, err = strconv.Atoi(string(decoded))
-		if err != nil || offset < 0 {
-			return nil, "", errors.New("analytics: invalid cursor")
-		}
-	}
-	statusPredicate := "TRUE"
-	switch filter.View {
-	case "ALL":
-		statusPredicate = "TRUE"
-	case "UNVERIFIED_PAID":
-		statusPredicate = "ps.status='Paid'"
-	case "FAILED":
-		statusPredicate = "ps.status='Failed'"
-	case "RETRIED":
-		statusPredicate = "(SELECT count(*) FROM payment_attempts x WHERE x.session_id=ps.id)>1"
+		cursorCreatedAt = &position.CreatedAt
+		cursorSessionID = position.SessionID
 	}
 	query := `SELECT ps.id,ps.source_key,t.source_key,ps.status,ps.amount_rial,ps.fee_amount_rial,ps.created_at,ps.expires_at,ps.verified_at,ps.settled_at,
-	 count(pa.id)::int,coalesce(bool_or(pa.status='InBank'),false) FROM payment_sessions ps JOIN terminals t ON t.id=ps.terminal_id LEFT JOIN payment_attempts pa ON pa.session_id=ps.id
-	 WHERE ps.merchant_id=$1 AND ps.created_at>=$2 AND ps.created_at<$3 AND ` + statusPredicate + `
-	 AND ($4='' OR t.source_key=$4)
-	 AND ($5='' OR
-	      ($5='NO_ATTEMPT' AND NOT EXISTS (SELECT 1 FROM payment_attempts fa WHERE fa.session_id=ps.id)) OR
-	      ($5='REACHED_BANK' AND EXISTS (SELECT 1 FROM payment_attempts fa WHERE fa.session_id=ps.id AND fa.status='InBank')) OR
-	      ($5='OTHER_OBSERVED' AND EXISTS (SELECT 1 FROM payment_attempts fa WHERE fa.session_id=ps.id) AND NOT EXISTS (SELECT 1 FROM payment_attempts fa WHERE fa.session_id=ps.id AND fa.status='InBank')))
-	 AND ($6='' OR ps.status=$6) AND ($7::bigint IS NULL OR ps.amount_rial >= $7) AND ($8::bigint IS NULL OR ps.amount_rial <= $8)
-	 GROUP BY ps.id,t.source_key ORDER BY ps.created_at DESC,ps.id DESC LIMIT $9 OFFSET $10`
-	rows, err := s.pool.Query(ctx, query, merchantID, period.StartAt, period.EndAt, filter.TerminalID, filter.FailureStage, filter.Status, filter.MinAmountRial, filter.MaxAmountRial, filter.Limit+1, offset)
+		 count(pa.id)::int,coalesce(bool_or(pa.status='InBank'),false) FROM payment_sessions ps JOIN terminals t ON t.id=ps.terminal_id LEFT JOIN payment_attempts pa ON pa.session_id=ps.id
+		 WHERE ` + detailFilterPredicate + `
+		 AND ($10::timestamptz IS NULL OR ps.created_at < $10 OR (ps.created_at=$10 AND ps.id<$11))
+		 GROUP BY ps.id,t.source_key ORDER BY ps.created_at DESC,ps.id DESC LIMIT $12`
+	rows, err := s.pool.Query(ctx, query, merchantID, period.StartAt, period.EndAt, filter.View, filter.TerminalID, filter.FailureStage, filter.Status, filter.MinAmountRial, filter.MaxAmountRial, cursorCreatedAt, cursorSessionID, filter.Limit+1)
 	if err != nil {
 		return nil, "", fmt.Errorf("querying session details: %w", err)
 	}
@@ -146,11 +152,53 @@ func (s *Service) Details(ctx context.Context, merchantID int64, period Period,
 	}
 	next := ""
 	if hasMore {
-		next = base64.RawURLEncoding.EncodeToString([]byte(strconv.Itoa(offset + filter.Limit)))
+		next, err = s.encodeDetailCursor(merchantID, period, filter, details[len(details)-1].CreatedAt, ids[len(ids)-1])
+		if err != nil {
+			return nil, "", fmt.Errorf("encoding detail cursor: %w", err)
+		}
 	}
 	return details, next, nil
 }

+func (s *Service) ExportDetails(ctx context.Context, merchantID int64, period Period, filter DetailFilter) ([]ExportDetail, error) {
+	filter.Cursor = ""
+	filter.Limit = 100
+	if err := filter.Validate(); err != nil {
+		return nil, err
+	}
+	query := `SELECT ps.id,ps.source_key,t.source_key,ps.status,ps.amount_rial,ps.fee_amount_rial,ps.created_at,ps.expires_at,ps.verified_at,ps.settled_at,
+		(SELECT count(*)::int FROM payment_attempts attempts WHERE attempts.session_id=ps.id),
+		EXISTS(SELECT 1 FROM payment_attempts attempts WHERE attempts.session_id=ps.id AND attempts.status='InBank'),
+		last_attempt.status,last_attempt.created_at
+		FROM payment_sessions ps JOIN terminals t ON t.id=ps.terminal_id
+		LEFT JOIN LATERAL (SELECT status,created_at FROM payment_attempts WHERE session_id=ps.id ORDER BY sequence DESC LIMIT 1) last_attempt ON true
+		WHERE ` + detailFilterPredicate + ` ORDER BY ps.created_at DESC,ps.id DESC LIMIT $10`
+	rows, err := s.pool.Query(ctx, query, merchantID, period.StartAt, period.EndAt, filter.View, filter.TerminalID, filter.FailureStage, filter.Status, filter.MinAmountRial, filter.MaxAmountRial, MaxExportRows+1)
+	if err != nil {
+		return nil, fmt.Errorf("querying session export: %w", err)
+	}
+	defer rows.Close()
+	items := make([]ExportDetail, 0, 1024)
+	for rows.Next() {
+		var id int64
+		var reachedBank bool
+		var item ExportDetail
+		if err := rows.Scan(&id, &item.SessionID, &item.TerminalID, &item.SessionStatus, &item.AmountRial, &item.FeeAmountRial, &item.CreatedAt, &item.ExpiresAt, &item.VerifiedAt, &item.SettledAt, &item.AttemptCount, &reachedBank, &item.LastAttemptStatus, &item.LastAttemptAt); err != nil {
+			return nil, fmt.Errorf("scanning session export: %w", err)
+		}
+		item.SessionID = "ses_" + item.SessionID
+		setFailureStage(&item.Detail, reachedBank)
+		items = append(items, item)
+		if len(items) > MaxExportRows {
+			return nil, ErrExportTooLarge
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterating session export: %w", err)
+	}
+	return items, nil
+}
+
 func (s *Service) Session(ctx context.Context, merchantID int64, sessionID string) (Detail, bool, error) {
 	sourceKey, valid := sessionSourceKey(sessionID)
 	if !valid {
--- a/internal/analytics/details_test.go
+++ b/internal/analytics/details_test.go
@@ -1,6 +1,11 @@
 package analytics

-import "testing"
+import (
+	"encoding/base64"
+	"errors"
+	"testing"
+	"time"
+)

 func TestDetailFilterValidate(t *testing.T) {
 	t.Parallel()
@@ -59,3 +64,53 @@ func TestTerminalDisplayName(t *testing.T) {
 		t.Fatalf("terminalDisplayName() = %q", got)
 	}
 }
+
+func TestDetailCursorRejectsMerchantAndFilterReplay(t *testing.T) {
+	t.Parallel()
+	service := &Service{cursorSecret: []byte("test-cursor-secret-that-is-at-least-32-bytes")}
+	period := Period{PeriodID: "jm-1405-03"}
+	minimum := int64(1000)
+	filter := DetailFilter{View: "ALL", TerminalID: "T318", Status: "Verified", MinAmountRial: &minimum, Limit: 25}
+	createdAt := time.Date(2026, time.June, 20, 8, 30, 0, 0, time.UTC)
+
+	cursor, err := service.encodeDetailCursor(42, period, filter, createdAt, 99)
+	if err != nil {
+		t.Fatal(err)
+	}
+	position, err := service.decodeDetailCursor(42, period, filter, cursor)
+	if err != nil {
+		t.Fatalf("decodeDetailCursor() error = %v", err)
+	}
+	if !position.CreatedAt.Equal(createdAt) || position.SessionID != 99 {
+		t.Fatalf("decodeDetailCursor() = %+v", position)
+	}
+
+	if _, err := service.decodeDetailCursor(43, period, filter, cursor); !errors.Is(err, ErrInvalidCursor) {
+		t.Fatalf("merchant replay error = %v, want ErrInvalidCursor", err)
+	}
+	changed := filter
+	changed.Status = "Paid"
+	if _, err := service.decodeDetailCursor(42, period, changed, cursor); !errors.Is(err, ErrInvalidCursor) {
+		t.Fatalf("filter replay error = %v, want ErrInvalidCursor", err)
+	}
+}
+
+func TestDetailCursorRejectsTampering(t *testing.T) {
+	t.Parallel()
+	service := &Service{cursorSecret: []byte("test-cursor-secret-that-is-at-least-32-bytes")}
+	period := Period{PeriodID: "jm-1405-03"}
+	filter := DetailFilter{View: "ALL", Limit: 25}
+	cursor, err := service.encodeDetailCursor(42, period, filter, time.Now().UTC(), 99)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sealed, err := base64.RawURLEncoding.DecodeString(cursor)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sealed[len(sealed)-1] ^= 1
+	tampered := base64.RawURLEncoding.EncodeToString(sealed)
+	if _, err := service.decodeDetailCursor(42, period, filter, tampered); !errors.Is(err, ErrInvalidCursor) {
+		t.Fatalf("tampered cursor error = %v, want ErrInvalidCursor", err)
+	}
+}
--- a/internal/analytics/service.go
+++ b/internal/analytics/service.go
@@ -49,9 +49,10 @@ type PaymentHistoryPoint struct {
 }

 type Service struct {
-	pool *pgxpool.Pool
-	mu   sync.RWMutex
-	runs map[string]storedRun
+	pool         *pgxpool.Pool
+	cursorSecret []byte
+	mu           sync.RWMutex
+	runs         map[string]storedRun
 }

 type storedRun struct {
@@ -59,7 +60,9 @@ type storedRun struct {
 	run        Run
 }

-func New(pool *pgxpool.Pool) *Service { return &Service{pool: pool, runs: map[string]storedRun{}} }
+func New(pool *pgxpool.Pool, cursorSecret string) *Service {
+	return &Service{pool: pool, cursorSecret: []byte(cursorSecret), runs: map[string]storedRun{}}
+}

 func (s *Service) Periods(ctx context.Context) ([]Period, error) {
 	var start, end time.Time
@@ -105,21 +108,27 @@ func (s *Service) Start(ctx context.Context, merchantID int64, period Period, te
 	run.Sections["overview"] = map[string]any{"sectionId": "overview", "state": "READY", "metadata": metadata, "data": metrics.overview(period, history)}
 	run.Sections["paymentPerformance"] = map[string]any{"sectionId": "payment_performance", "state": "READY", "metadata": metadata, "data": metrics.performance()}
 	run.Sections["fees"] = map[string]any{"sectionId": "fees", "state": "READY", "metadata": metadata, "data": metrics.fees(period)}
-	peerCount, peerSessions, err := s.peerEligibility(ctx, merchantID, period)
-	if err != nil {
-		return Run{}, err
-	}
-	if peerCount < 10 || peerSessions < 1000 {
+	if !peerComparisonSupported(terminalID) {
 		run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "SUPPRESSED", "metadata": metadata,
-			"reason": "INSUFFICIENT_COHORT", "eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions,
+			"reason": "TERMINAL_SCOPE_UNSUPPORTED", "eligiblePeerCount": 0, "combinedPeerSessionCount": 0,
 			"requiredPeerCount": 10, "requiredPeerSessionCount": 1000}
 	} else {
-		peerMetrics, err := s.peerMetrics(ctx, merchantID, period)
+		peerCount, peerSessions, err := s.peerEligibility(ctx, merchantID, period)
 		if err != nil {
 			return Run{}, err
 		}
-		run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "READY", "metadata": metadata,
-			"data": map[string]any{"eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions, "metrics": peerMetrics}}
+		if peerCount < 10 || peerSessions < 1000 {
+			run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "SUPPRESSED", "metadata": metadata,
+				"reason": "INSUFFICIENT_COHORT", "eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions,
+				"requiredPeerCount": 10, "requiredPeerSessionCount": 1000}
+		} else {
+			peerMetrics, err := s.peerMetrics(ctx, merchantID, period)
+			if err != nil {
+				return Run{}, err
+			}
+			run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "READY", "metadata": metadata,
+				"data": map[string]any{"eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions, "metrics": peerMetrics}}
+		}
 	}
 	s.mu.Lock()
 	s.runs[key] = storedRun{merchantID: merchantID, run: run}
@@ -128,6 +137,8 @@ func (s *Service) Start(ctx context.Context, merchantID int64, period Period, te
 	return run, nil
 }

+func peerComparisonSupported(terminalID string) bool { return terminalID == "" }
+
 func (s *Service) Run(merchantID int64, runID string) (Run, bool) {
 	s.mu.RLock()
 	stored, ok := s.runs[runID]
--- a/internal/analytics/service_test.go
+++ b/internal/analytics/service_test.go
@@ -0,0 +1,13 @@
+package analytics
+
+import "testing"
+
+func TestPeerComparisonSupported(t *testing.T) {
+	t.Parallel()
+	if peerComparisonSupported("T318") {
+		t.Fatal("peerComparisonSupported() = true for terminal scope")
+	}
+	if !peerComparisonSupported("") {
+		t.Fatal("peerComparisonSupported() = false for merchant scope")
+	}
+}