// Package errors provides a unified error hierarchy for the streaming pipeline. // It categorizes errors and provides retry semantics for resilient operation handling. package errors import ( "encoding/json" "fmt" ) // ErrorCategory represents the category of an error for handling decisions. type ErrorCategory int const ( // CategoryUnknown indicates an uncategorized error CategoryUnknown ErrorCategory = iota // CategoryAuth indicates an authentication/authorization error CategoryAuth // CategoryRateLimit indicates a rate limiting error CategoryRateLimit // CategoryValidation indicates a request validation error CategoryValidation // CategoryUpstream indicates an upstream provider error CategoryUpstream // CategoryInternal indicates an internal system error CategoryInternal // CategoryNetwork indicates a network error CategoryNetwork // CategoryTimeout indicates a timeout error CategoryTimeout ) func (c ErrorCategory) String() string { switch c { case CategoryAuth: return "auth" case CategoryRateLimit: return "rate_limit" case CategoryValidation: return "validation" case CategoryUpstream: return "upstream" case CategoryInternal: return "internal" case CategoryNetwork: return "network" case CategoryTimeout: return "timeout" default: return "unknown" } } // PipelineError is the unified error type for the streaming pipeline. // It provides categorization, retry semantics, and context propagation. type PipelineError struct { Category ErrorCategory `json:"category"` StatusCode int `json:"status_code"` Message string `json:"message"` Cause error `json:"-"` Retryable bool `json:"retryable"` Context map[string]interface{} `json:"context,omitempty"` } // Error implements the error interface. func (e *PipelineError) Error() string { if e.Cause != nil { return fmt.Sprintf("[%s] %s: %v", e.Category, e.Message, e.Cause) } return fmt.Sprintf("[%s] %s", e.Category, e.Message) } // Unwrap returns the wrapped error for error chain inspection. func (e *PipelineError) Unwrap() error { return e.Cause } // MarshalJSON implements json.Marshaler for API responses. func (e *PipelineError) MarshalJSON() ([]byte, error) { type Alias PipelineError return json.Marshal(&struct { *Alias Cause string `json:"cause,omitempty"` }{ Alias: (*Alias)(e), Cause: func() string { if e.Cause != nil { return e.Cause.Error() } return "" }(), }) } // WithContext adds context information to the error. func (e *PipelineError) WithContext(key string, value interface{}) *PipelineError { if e.Context == nil { e.Context = make(map[string]interface{}) } e.Context[key] = value return e } // IsRetryable returns true if the error is retryable. func (e *PipelineError) IsRetryable() bool { return e.Retryable } // New creates a new PipelineError with the given category and message. func New(category ErrorCategory, message string) *PipelineError { return &PipelineError{ Category: category, Message: message, StatusCode: defaultStatusCode(category), Retryable: defaultRetryable(category), } } // NewWithCause creates a new PipelineError with a cause. func NewWithCause(category ErrorCategory, message string, cause error) *PipelineError { return &PipelineError{ Category: category, Message: message, Cause: cause, StatusCode: defaultStatusCode(category), Retryable: defaultRetryable(category), } } // Wrap wraps an existing error as a PipelineError. func Wrap(err error, category ErrorCategory, message string) *PipelineError { if err == nil { return nil } // If already a PipelineError, just update fields if pe, ok := err.(*PipelineError); ok { if message != "" { pe.Message = message } if category != CategoryUnknown { pe.Category = category } return pe } return &PipelineError{ Category: category, Message: message, Cause: err, StatusCode: defaultStatusCode(category), Retryable: defaultRetryable(category), } } // Authentication errors // AuthError creates an authentication error. func AuthError(message string) *PipelineError { return New(CategoryAuth, message) } // AuthErrorWithCause creates an authentication error with cause. func AuthErrorWithCause(message string, cause error) *PipelineError { return NewWithCause(CategoryAuth, message, cause) } // Rate limit errors // RateLimitError creates a rate limit error. func RateLimitError(message string) *PipelineError { return New(CategoryRateLimit, message) } // RateLimitErrorWithRetryAfter creates a rate limit error with retry-after information. func RateLimitErrorWithRetryAfter(message string, retryAfterSeconds int) *PipelineError { e := New(CategoryRateLimit, message) e.Retryable = true return e.WithContext("retry_after_seconds", retryAfterSeconds) } // Validation errors // ValidationError creates a validation error. func ValidationError(message string) *PipelineError { return New(CategoryValidation, message) } // ValidationErrorWithField creates a validation error with field information. func ValidationErrorWithField(message, field string) *PipelineError { return New(CategoryValidation, message).WithContext("field", field) } // Upstream errors // UpstreamError creates an upstream provider error. func UpstreamError(message string) *PipelineError { return New(CategoryUpstream, message) } // UpstreamErrorWithCause creates an upstream error with cause. func UpstreamErrorWithCause(message string, cause error) *PipelineError { return NewWithCause(CategoryUpstream, message, cause) } // Internal errors // InternalError creates an internal error. func InternalError(message string) *PipelineError { return New(CategoryInternal, message) } // InternalErrorWithCause creates an internal error with cause. func InternalErrorWithCause(message string, cause error) *PipelineError { return NewWithCause(CategoryInternal, message, cause) } // Network errors // NetworkError creates a network error. func NetworkError(message string) *PipelineError { return New(CategoryNetwork, message) } // NetworkErrorWithCause creates a network error with cause. func NetworkErrorWithCause(message string, cause error) *PipelineError { return NewWithCause(CategoryNetwork, message, cause) } // Timeout errors // TimeoutError creates a timeout error. func TimeoutError(message string) *PipelineError { return New(CategoryTimeout, message) } // Helper functions func defaultStatusCode(category ErrorCategory) int { switch category { case CategoryAuth: return 401 case CategoryRateLimit: return 429 case CategoryValidation: return 400 case CategoryUpstream: return 502 case CategoryInternal: return 500 case CategoryNetwork: return 503 case CategoryTimeout: return 504 default: return 500 } } func defaultRetryable(category ErrorCategory) bool { switch category { case CategoryRateLimit, CategoryNetwork, CategoryTimeout: return true case CategoryAuth, CategoryValidation: return false case CategoryUpstream: return true case CategoryInternal: return false default: return false } } // IsPipelineError checks if an error is a PipelineError. func IsPipelineError(err error) (*PipelineError, bool) { if err == nil { return nil, false } if pe, ok := err.(*PipelineError); ok { return pe, true } return nil, false } // IsCategory checks if an error belongs to a specific category. func IsCategory(err error, category ErrorCategory) bool { if pe, ok := IsPipelineError(err); ok { return pe.Category == category } return false } // GetCategory returns the category of an error, or CategoryUnknown. func GetCategory(err error) ErrorCategory { if pe, ok := IsPipelineError(err); ok { return pe.Category } return CategoryUnknown } // IsRetryableError checks if an error is retryable. func IsRetryableError(err error) bool { if pe, ok := IsPipelineError(err); ok { return pe.Retryable } // Default: unknown errors are not retryable return false } // ErrorResponse is the JSON API error response structure. type ErrorResponse struct { Error string `json:"error"` Category string `json:"category"` StatusCode int `json:"status_code"` Retryable bool `json:"retryable"` Context map[string]interface{} `json:"context,omitempty"` } // ToResponse converts a PipelineError to an ErrorResponse. func (e *PipelineError) ToResponse() ErrorResponse { return ErrorResponse{ Error: e.Message, Category: e.Category.String(), StatusCode: e.StatusCode, Retryable: e.Retryable, Context: e.Context, } }