Extract handler store layer, fix N+1 test result inserts

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented here.
 
 ## [Unreleased]
 
+### Changed
+
+- **Store layer extraction**: All HTTP handlers now use store interfaces instead of embedding `*db.Pool` directly. Store interfaces are defined on the handler side and implemented in `internal/store/`, making handlers testable without a running database and centralizing SQL query knowledge. Affected handlers: auth, teams, analytics, executions, reports, admin, invitations, oauth.
+
+- **Bulk test result inserts**: Report ingestion now uses `pgx.Batch` to insert test results in bulk instead of one query per result. This eliminates the N+1 insert pattern that caused 1000+ round-trips for large reports.
+
 ### Fixed
 
 - **IDOR vulnerability in invitation handlers**: `Create`, `List`, and `Revoke` invitation endpoints (`POST/GET/DELETE /api/v1/teams/{teamID}/invitations`) now verify that the authenticated user's team matches the URL `teamID` before checking role permissions. Previously, any maintainer or owner could list, create, or revoke invitations for any team regardless of membership.

README.md

@@ -497,7 +497,7 @@ internal/
   db/                 # Database pool, migrations
   handler/            # HTTP handlers (reports, executions, teams, admin, etc.)
   server/             # Router and middleware setup
-  store/              # Data access (audit, webhooks, quality gates)
+  store/              # Data access layer (store interfaces + implementations)
   github/             # GitHub commit status client
   llm/                # LLM provider abstraction (Anthropic, OpenAI, mock)
   mail/               # Email sender interface and SMTP implementation

internal/handler/admin.go

@@ -5,15 +5,14 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/scaledtest/scaledtest/internal/db"
-	"github.com/scaledtest/scaledtest/internal/model"
+
 	"github.com/scaledtest/scaledtest/internal/store"
 )
 
 // AdminHandler handles admin-only endpoints.
 type AdminHandler struct {
 	AuditStore *store.AuditStore
-	DB         *db.Pool
+	AdminStore adminStore
 }
 
 // ListAuditLog handles GET /api/v1/admin/audit-log.
@@ -66,46 +65,18 @@ func (h *AdminHandler) ListAuditLog(w http.ResponseWriter, r *http.Request) {
 
 // ListUsers handles GET /api/v1/admin/users.
 func (h *AdminHandler) ListUsers(w http.ResponseWriter, r *http.Request) {
-	if h.DB == nil {
+	if h.AdminStore == nil {
 		Error(w, http.StatusServiceUnavailable, "database not configured")
 		return
 	}
 
 	limit, offset := parsePagination(r)
 
-	rows, err := h.DB.Query(r.Context(),
-		`SELECT id, email, display_name, role, created_at, updated_at
-		 FROM users
-		 ORDER BY created_at DESC
-		 LIMIT $1 OFFSET $2`,
-		limit, offset)
+	users, total, err := h.AdminStore.ListUsers(r.Context(), limit, offset)
 	if err != nil {
 		Error(w, http.StatusInternalServerError, "failed to query users")
 		return
 	}
-	defer rows.Close()
-
-	users := []model.User{}
-	for rows.Next() {
-		var u model.User
-		if err := rows.Scan(&u.ID, &u.Email, &u.DisplayName, &u.Role, &u.CreatedAt, &u.UpdatedAt); err != nil {
-			Error(w, http.StatusInternalServerError, "failed to scan user")
-			return
-		}
-		users = append(users, u)
-	}
-	if err := rows.Err(); err != nil {
-		Error(w, http.StatusInternalServerError, "failed to iterate users")
-		return
-	}
-
-	var total int
-	err = h.DB.QueryRow(r.Context(), `SELECT COUNT(*) FROM users`).Scan(&total)
-	if err != nil {
-		Error(w, http.StatusInternalServerError, "failed to count users")
-		return
-	}
-
 	JSON(w, http.StatusOK, map[string]interface{}{
 		"users": users,
 		"total": total,

internal/handler/analytics.go

@@ -9,12 +9,11 @@ import (
 
 	"github.com/scaledtest/scaledtest/internal/analytics"
 	"github.com/scaledtest/scaledtest/internal/auth"
-	"github.com/scaledtest/scaledtest/internal/db"
 )
 
 // AnalyticsHandler handles analytics endpoints.
 type AnalyticsHandler struct {
-	DB *db.Pool
+	AnalyticsStore analyticsStore
 }
 
 // Trends handles GET /api/v1/analytics/trends.
@@ -26,57 +25,30 @@ func (h *AnalyticsHandler) Trends(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if h.DB == nil {
+	if h.AnalyticsStore == nil {
 		Error(w, http.StatusServiceUnavailable, "database not configured")
 		return
 	}
 
 	q := parseTrendQuery(r, claims.TeamID)
 
-	query := `
-		SELECT
-			time_bucket($1::interval, created_at) AS bucket,
-			count(*) AS total,
-			count(*) FILTER (WHERE status = 'passed') AS passed,
-			count(*) FILTER (WHERE status = 'failed') AS failed,
-			count(*) FILTER (WHERE status = 'skipped') AS skipped
-		FROM test_results
-		WHERE team_id = $2
-			AND created_at >= $3
-			AND created_at <= $4
-		GROUP BY bucket
-		ORDER BY bucket
-	`
-
-	rows, err := h.DB.Query(r.Context(), query, q.GroupBy, q.TeamID, q.StartDate, q.EndDate)
+	rows, err := h.AnalyticsStore.QueryTrends(r.Context(), q.GroupBy, q.TeamID, q.StartDate, q.EndDate)
 	if err != nil {
 		log.Error().Err(err).Msg("analytics: trends query failed")
 		Error(w, http.StatusInternalServerError, "query failed")
 		return
 	}
-	defer rows.Close()
-
-	var trends []analytics.TrendPoint
-	for rows.Next() {
-		var tp analytics.TrendPoint
-		if err := rows.Scan(&tp.Date, &tp.Total, &tp.Passed, &tp.Failed, &tp.Skipped); err != nil {
-			log.Error().Err(err).Msg("analytics: trends scan failed")
-			Error(w, http.StatusInternalServerError, "query failed")
-			return
+	trends := make([]analytics.TrendPoint, len(rows))
+	for i, row := range rows {
+		trends[i] = analytics.TrendPoint{
+			Date:     row.Date,
+			Total:    row.Total,
+			Passed:   row.Passed,
+			Failed:   row.Failed,
+			Skipped:  row.Skipped,
+			PassRate: row.PassRate,
 		}
-		tp.PassRate = analytics.ComputePassRate(tp.Passed, tp.Total)
-		trends = append(trends, tp)
 	}
-	if err := rows.Err(); err != nil {
-		log.Error().Err(err).Msg("analytics: trends iteration failed")
-		Error(w, http.StatusInternalServerError, "query failed")
-		return
-	}
-
-	if trends == nil {
-		trends = []analytics.TrendPoint{}
-	}
-
 	JSON(w, http.StatusOK, map[string]interface{}{
 		"trends": trends,
 	})
@@ -91,64 +63,38 @@ func (h *AnalyticsHandler) FlakyTests(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if h.DB == nil {
+	if h.AnalyticsStore == nil {
 		Error(w, http.StatusServiceUnavailable, "database not configured")
 		return
 	}
 
 	q := parseFlakyQuery(r, claims.TeamID)
 	cutoff := time.Now().Add(-q.Window)
 
-	// Query: for each test, get ordered statuses, then compute flakiness in Go.
-	// This avoids complex SQL window functions and uses the analytics.DetectFlaky helper.
-	query := `
-		SELECT
-			name,
-			COALESCE(suite, '') AS suite,
-			COALESCE(file_path, '') AS file_path,
-			array_agg(status ORDER BY created_at) AS statuses,
-			(array_agg(status ORDER BY created_at DESC))[1] AS last_status,
-			count(*) AS total_runs
-		FROM test_results
-		WHERE team_id = $1
-			AND created_at >= $2
-		GROUP BY name, suite, file_path
-		HAVING count(*) >= $3
-		ORDER BY name
-	`
-
-	rows, err := h.DB.Query(r.Context(), query, q.TeamID, cutoff, q.MinRuns)
+	rows, err := h.AnalyticsStore.QueryFlakyTests(r.Context(), claims.TeamID, cutoff, q.MinRuns)
 	if err != nil {
 		log.Error().Err(err).Msg("analytics: flaky query failed")
 		Error(w, http.StatusInternalServerError, "query failed")
 		return
 	}
-	defer rows.Close()
 
 	var flaky []analytics.FlakyTest
-	for rows.Next() {
-		var ft analytics.FlakyTest
-		var statuses []string
-		if err := rows.Scan(&ft.Name, &ft.Suite, &ft.FilePath, &statuses, &ft.LastStatus, &ft.TotalRuns); err != nil {
-			log.Error().Err(err).Msg("analytics: flaky scan failed")
-			Error(w, http.StatusInternalServerError, "query failed")
-			return
+	for _, fr := range rows {
+		ft := analytics.FlakyTest{
+			Name:       fr.Name,
+			Suite:      fr.Suite,
+			FilePath:   fr.FilePath,
+			LastStatus: fr.LastStatus,
+			TotalRuns:  fr.TotalRuns,
 		}
-
-		ft.FlipCount, ft.FlipRate = analytics.DetectFlaky(statuses)
+		ft.FlipCount, ft.FlipRate = analytics.DetectFlaky(fr.Statuses)
 		if ft.FlipCount > 0 {
 			flaky = append(flaky, ft)
 		}
-
 		if q.Limit > 0 && len(flaky) >= q.Limit {
 			break
 		}
 	}
-	if err := rows.Err(); err != nil {
-		log.Error().Err(err).Msg("analytics: flaky iteration failed")
-		Error(w, http.StatusInternalServerError, "query failed")
-		return
-	}
 
 	if flaky == nil {
 		flaky = []analytics.FlakyTest{}
@@ -168,63 +114,34 @@ func (h *AnalyticsHandler) ErrorAnalysis(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	if h.DB == nil {
+	if h.AnalyticsStore == nil {
 		Error(w, http.StatusServiceUnavailable, "database not configured")
 		return
 	}
 
 	start, end := parseDateRange(r)
 	limit := parseIntParam(r, "limit", 20)
 
-	query := `
-		SELECT
-			message,
-			count(*) AS count,
-			array_agg(DISTINCT name) AS test_names,
-			min(created_at) AS first_seen,
-			max(created_at) AS last_seen
-		FROM test_results
-		WHERE team_id = $1
-			AND status = 'failed'
-			AND message IS NOT NULL
-			AND message != ''
-			AND created_at >= $2
-			AND created_at <= $3
-		GROUP BY message
-		ORDER BY count DESC
-		LIMIT $4
-	`
-
-	rows, err := h.DB.Query(r.Context(), query, claims.TeamID, start, end, limit)
+	clusters, err := h.AnalyticsStore.QueryErrorClusters(r.Context(), claims.TeamID, start, end, limit)
 	if err != nil {
 		log.Error().Err(err).Msg("analytics: error analysis query failed")
 		Error(w, http.StatusInternalServerError, "query failed")
 		return
 	}
-	defer rows.Close()
-
-	var clusters []analytics.ErrorCluster
-	for rows.Next() {
-		var ec analytics.ErrorCluster
-		if err := rows.Scan(&ec.Message, &ec.Count, &ec.TestNames, &ec.FirstSeen, &ec.LastSeen); err != nil {
-			log.Error().Err(err).Msg("analytics: error analysis scan failed")
-			Error(w, http.StatusInternalServerError, "query failed")
-			return
-		}
-		clusters = append(clusters, ec)
-	}
-	if err := rows.Err(); err != nil {
-		log.Error().Err(err).Msg("analytics: error analysis iteration failed")
-		Error(w, http.StatusInternalServerError, "query failed")
-		return
-	}
 
-	if clusters == nil {
-		clusters = []analytics.ErrorCluster{}
+	result := make([]analytics.ErrorCluster, len(clusters))
+	for i, ec := range clusters {
+		result[i] = analytics.ErrorCluster{
+			Message:   ec.Message,
+			Count:     ec.Count,
+			TestNames: ec.TestNames,
+			FirstSeen: ec.FirstSeen,
+			LastSeen:  ec.LastSeen,
+		}
 	}
 
 	JSON(w, http.StatusOK, map[string]interface{}{
-		"errors": clusters,
+		"errors": result,
 	})
 }
 
@@ -237,46 +154,24 @@ func (h *AnalyticsHandler) DurationDistribution(w http.ResponseWriter, r *http.R
 		return
 	}
 
-	if h.DB == nil {
+	if h.AnalyticsStore == nil {
 		Error(w, http.StatusServiceUnavailable, "database not configured")
 		return
 	}
 
 	start, end := parseDateRange(r)
 
-	// Build histogram buckets
 	buckets := analytics.DefaultDurationBuckets()
-	bucketQuery := `
-		SELECT duration_ms
-		FROM test_results
-		WHERE team_id = $1
-			AND created_at >= $2
-			AND created_at <= $3
-	`
-
-	rows, err := h.DB.Query(r.Context(), bucketQuery, claims.TeamID, start, end)
+	durations, err := h.AnalyticsStore.QueryDurationBuckets(r.Context(), claims.TeamID, start, end)
 	if err != nil {
 		log.Error().Err(err).Msg("analytics: duration bucket query failed")
 		Error(w, http.StatusInternalServerError, "query failed")
 		return
 	}
-	defer rows.Close()
-
-	for rows.Next() {
-		var ms int64
-		if err := rows.Scan(&ms); err != nil {
-			log.Error().Err(err).Msg("analytics: duration bucket scan failed")
-			Error(w, http.StatusInternalServerError, "query failed")
-			return
-		}
+	for _, ms := range durations {
 		idx := analytics.BucketDuration(ms, buckets)
 		buckets[idx].Count++
 	}
-	if err := rows.Err(); err != nil {
-		log.Error().Err(err).Msg("analytics: duration bucket iteration failed")
-		Error(w, http.StatusInternalServerError, "query failed")
-		return
-	}
 
 	JSON(w, http.StatusOK, map[string]interface{}{
 		"distribution": buckets,