Edge case fixes: race conditions, resource leaks, and input validation

- Fix race condition in batch.go with mutex protection for results slice
- Fix potential channel deadlock in ffmpeg_version.go SSE streaming
- Add Close() method to GCS provider to prevent connection pool leaks
- Add nil checks to handler constructors (batch, ffmpeg_version, api_keys)
- Add ExpiresIn validation (1-365 days) to API key creation
- Extract isAdmin/requireAdmin helpers to reduce code duplication

Author: alokemajumder

internal/handlers/api_keys.go

@@ -17,17 +17,43 @@ type APIKeyHandler struct {
 
 // NewAPIKeyHandler creates a new API key handler
 func NewAPIKeyHandler(rotationService *services.SecretRotationService, logger zerolog.Logger) *APIKeyHandler {
+	if rotationService == nil {
+		panic("rotationService cannot be nil")
+	}
 	return &APIKeyHandler{
 		rotationService: rotationService,
 		logger:          logger,
 	}
 }
 
+// isAdmin checks if the current user has admin role
+func isAdmin(c *gin.Context) bool {
+	roles := c.GetStringSlice("roles")
+	for _, role := range roles {
+		if role == "admin" {
+			return true
+		}
+	}
+	return false
+}
+
+// requireAdmin returns an error response if the user is not an admin, returns true if admin
+func requireAdmin(c *gin.Context) bool {
+	if !isAdmin(c) {
+		c.JSON(http.StatusForbidden, gin.H{
+			"error":   "Forbidden",
+			"message": "Admin role required",
+		})
+		return false
+	}
+	return true
+}
+
 // CreateAPIKeyRequest represents the request to create a new API key
 type CreateAPIKeyRequest struct {
 	Name        string   `json:"name" binding:"required,min=3,max=100"`
 	Permissions []string `json:"permissions"`
-	ExpiresIn   int      `json:"expires_in_days,omitempty"` // Optional, defaults to 90 days
+	ExpiresIn   int      `json:"expires_in_days,omitempty" binding:"omitempty,min=1,max=365"` // Optional, defaults to 90 days, max 1 year
 }
 
 // RotateAPIKeyRequest represents the request to rotate an API key
@@ -183,21 +209,7 @@ func (h *APIKeyHandler) RotateAPIKey(c *gin.Context) {
 
 // RotateJWTSecret rotates the JWT signing secret (admin only)
 func (h *APIKeyHandler) RotateJWTSecret(c *gin.Context) {
-	// Check admin role
-	roles := c.GetStringSlice("roles")
-	isAdmin := false
-	for _, role := range roles {
-		if role == "admin" {
-			isAdmin = true
-			break
-		}
-	}
-
-	if !isAdmin {
-		c.JSON(http.StatusForbidden, gin.H{
-			"error":   "Forbidden",
-			"message": "Admin role required",
-		})
+	if !requireAdmin(c) {
 		return
 	}
 
@@ -242,21 +254,7 @@ func (h *APIKeyHandler) UpdateRateLimits(c *gin.Context) {
 		return
 	}
 
-	// Check admin role
-	roles := c.GetStringSlice("roles")
-	isAdmin := false
-	for _, role := range roles {
-		if role == "admin" {
-			isAdmin = true
-			break
-		}
-	}
-
-	if !isAdmin {
-		c.JSON(http.StatusForbidden, gin.H{
-			"error":   "Forbidden",
-			"message": "Admin role required",
-		})
+	if !requireAdmin(c) {
 		return
 	}
 
@@ -319,21 +317,7 @@ func (h *APIKeyHandler) UpdateRateLimits(c *gin.Context) {
 
 // CheckRotationStatus checks which secrets are due for rotation (admin only)
 func (h *APIKeyHandler) CheckRotationStatus(c *gin.Context) {
-	// Check admin role
-	roles := c.GetStringSlice("roles")
-	isAdmin := false
-	for _, role := range roles {
-		if role == "admin" {
-			isAdmin = true
-			break
-		}
-	}
-
-	if !isAdmin {
-		c.JSON(http.StatusForbidden, gin.H{
-			"error":   "Forbidden",
-			"message": "Admin role required",
-		})
+	if !requireAdmin(c) {
 		return
 	}
 
@@ -361,21 +345,7 @@ func (h *APIKeyHandler) CheckRotationStatus(c *gin.Context) {
 
 // CleanupExpiredKeys removes expired keys past their grace period (admin only)
 func (h *APIKeyHandler) CleanupExpiredKeys(c *gin.Context) {
-	// Check admin role
-	roles := c.GetStringSlice("roles")
-	isAdmin := false
-	for _, role := range roles {
-		if role == "admin" {
-			isAdmin = true
-			break
-		}
-	}
-
-	if !isAdmin {
-		c.JSON(http.StatusForbidden, gin.H{
-			"error":   "Forbidden",
-			"message": "Admin role required",
-		})
+	if !requireAdmin(c) {
 		return
 	}
 

internal/handlers/batch.go

@@ -26,6 +26,9 @@ type BatchHandler struct {
 
 // NewBatchHandler creates a new batch handler
 func NewBatchHandler(analysisService *services.AnalysisService, logger zerolog.Logger) *BatchHandler {
+	if analysisService == nil {
+		panic("analysisService cannot be nil")
+	}
 	return &BatchHandler{
 		analysisService: analysisService,
 		maxBatchSize:    100, // Default max batch size
@@ -357,6 +360,7 @@ func (h *BatchHandler) processBatchAsync(batchID uuid.UUID, req BatchAnalysisReq
 	// Process files concurrently with limited concurrency
 	sem := make(chan struct{}, 5) // Limit to 5 concurrent analyses
 	var wg sync.WaitGroup
+	var resultsMutex sync.Mutex
 	results := make([]BatchResultItem, len(req.Files))
 
 	for i, file := range req.Files {
@@ -382,7 +386,9 @@ func (h *BatchHandler) processBatchAsync(batchID uuid.UUID, req BatchAnalysisReq
 
 			if cancelled {
 				result.Status = "cancelled"
+				resultsMutex.Lock()
 				results[index] = result
+				resultsMutex.Unlock()
 				return
 			}
 
@@ -400,8 +406,12 @@ func (h *BatchHandler) processBatchAsync(batchID uuid.UUID, req BatchAnalysisReq
 			if err != nil {
 				result.Status = "failed"
 				result.Error = err.Error()
+				resultsMutex.Lock()
 				results[index] = result
-				h.updateBatchProgress(batchID, results)
+				resultsCopy := make([]BatchResultItem, len(results))
+				copy(resultsCopy, results)
+				resultsMutex.Unlock()
+				h.updateBatchProgress(batchID, resultsCopy)
 				return
 			}
 
@@ -420,8 +430,12 @@ func (h *BatchHandler) processBatchAsync(batchID uuid.UUID, req BatchAnalysisReq
 				result.Status = "completed"
 			}
 
+			resultsMutex.Lock()
 			results[index] = result
-			h.updateBatchProgress(batchID, results)
+			resultsCopy := make([]BatchResultItem, len(results))
+			copy(resultsCopy, results)
+			resultsMutex.Unlock()
+			h.updateBatchProgress(batchID, resultsCopy)
 		}(i, file)
 	}
 

internal/handlers/ffmpeg_version.go

@@ -17,6 +17,9 @@ type FFmpegVersionHandler struct {
 
 // NewFFmpegVersionHandler creates a new FFmpeg version handler
 func NewFFmpegVersionHandler(updater *services.FFmpegUpdater, logger zerolog.Logger) *FFmpegVersionHandler {
+	if updater == nil {
+		panic("updater cannot be nil")
+	}
 	return &FFmpegVersionHandler{
 		updater: updater,
 		logger:  logger,
@@ -121,13 +124,18 @@ func (h *FFmpegVersionHandler) UpdateFFmpeg(c *gin.Context) {
 
 	// Perform the update
 	progressChan := make(chan int, 100)
-	errorChan := make(chan error, 1)
+	doneChan := make(chan error, 1)
 
 	go func() {
 		err := h.updater.DownloadUpdate(ctx, updateInfo.Available, func(percent int) {
-			progressChan <- percent
+			select {
+			case progressChan <- percent:
+			default:
+				// Channel full, skip this update
+			}
 		})
-		errorChan <- err
+		doneChan <- err
+		close(doneChan)
 		close(progressChan)
 	}()
 
@@ -140,11 +148,17 @@ func (h *FFmpegVersionHandler) UpdateFFmpeg(c *gin.Context) {
 		select {
 		case progress, ok := <-progressChan:
 			if !ok {
-				// Channel closed, check for error
-				if err := <-errorChan; err != nil {
-					h.logger.Error().Err(err).Msg("FFmpeg update failed")
-					c.SSEvent("error", gin.H{"data": err.Error()})
-				} else {
+				// Progress channel closed, get result from done channel
+				select {
+				case err := <-doneChan:
+					if err != nil {
+						h.logger.Error().Err(err).Msg("FFmpeg update failed")
+						c.SSEvent("error", gin.H{"data": err.Error()})
+					} else {
+						c.SSEvent("complete", gin.H{"data": "FFmpeg updated successfully"})
+					}
+				default:
+					// Already consumed or not available
 					c.SSEvent("complete", gin.H{"data": "FFmpeg updated successfully"})
 				}
 				return false

internal/storage/gcs.go

@@ -95,3 +95,11 @@ func (g *GCSProvider) GetSignedURL(ctx context.Context, key string, expiration i
 
 	return url, nil
 }
+
+// Close closes the GCS client and releases resources
+func (g *GCSProvider) Close() error {
+	if g.client != nil {
+		return g.client.Close()
+	}
+	return nil
+}