diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 5a1db4c..c548397 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -22,9 +22,6 @@ jobs: runs-on: docker container: image: golang:1.25-alpine - # Advisory while the pre-existing lint backlog (~95 findings) is worked - # off; flip to blocking by removing continue-on-error once clean. - continue-on-error: true steps: - name: Checkout run: | diff --git a/api/grpc/handler.go b/api/grpc/handler.go index 0939560..2bfa624 100644 --- a/api/grpc/handler.go +++ b/api/grpc/handler.go @@ -1,8 +1,12 @@ +// Package grpc implements the gRPC transport for the notifier service, +// translating between the generated protobuf types and internal domain +// types. package grpc import ( "context" "fmt" + "math" "github.com/google/uuid" pb "github.com/igodwin/notifier/api/grpc/pb" @@ -29,7 +33,7 @@ func NewNotifierHandler(svc domain.NotificationService, logger *logging.Logger) } // HealthCheck verifies the service is operational -func (h *NotifierHandler) HealthCheck(ctx context.Context, req *pb.HealthCheckRequest) (*pb.HealthCheckResponse, error) { +func (h *NotifierHandler) HealthCheck(_ context.Context, _ *pb.HealthCheckRequest) (*pb.HealthCheckResponse, error) { // TODO: Implement proper health check logic return &pb.HealthCheckResponse{ Healthy: true, @@ -56,7 +60,7 @@ func (h *NotifierHandler) SendNotification(ctx context.Context, req *pb.SendNoti } // Convert content type, defaulting to text - contentType := convertProtoContentTypeToDomain(req.ContentType) + contentType := convertProtoContentTypeToDomain(req.ContentType) //nolint:staticcheck // deprecated content_type field still honored for backward compatibility // Build notification notification := &domain.Notification{ @@ -205,7 +209,7 @@ func (h *NotifierHandler) RetryNotification(ctx context.Context, req *pb.RetryNo } // GetStats returns notification statistics -func (h *NotifierHandler) GetStats(ctx context.Context, req *pb.GetStatsRequest) (*pb.GetStatsResponse, error) { +func (h *NotifierHandler) GetStats(ctx context.Context, _ *pb.GetStatsRequest) (*pb.GetStatsResponse, error) { stats, err := h.service.GetStats(ctx) if err != nil { return nil, err @@ -222,7 +226,7 @@ func (h *NotifierHandler) GetStats(ctx context.Context, req *pb.GetStatsRequest) } // GetNotifiers returns information about available notifiers -func (h *NotifierHandler) GetNotifiers(ctx context.Context, req *pb.GetNotifiersRequest) (*pb.GetNotifiersResponse, error) { +func (h *NotifierHandler) GetNotifiers(ctx context.Context, _ *pb.GetNotifiersRequest) (*pb.GetNotifiersResponse, error) { h.logger.Infof("gRPC: Received request for available notifiers") notifiers, err := h.service.GetNotifiers(ctx) @@ -248,6 +252,18 @@ func (h *NotifierHandler) GetNotifiers(ctx context.Context, req *pb.GetNotifiers // Helper functions to convert between proto and domain types +// clampInt32 narrows an int to int32, saturating at the int32 bounds +// instead of silently wrapping when the domain value is out of range. +func clampInt32(v int) int32 { + if v > math.MaxInt32 { + return math.MaxInt32 + } + if v < math.MinInt32 { + return math.MinInt32 + } + return int32(v) +} + // convertStringMapToInterface converts proto's map[string]string to domain's map[string]interface{} func convertStringMapToInterface(m map[string]string) map[string]interface{} { if m == nil { @@ -315,17 +331,6 @@ func convertProtoContentTypeToDomain(protoType pb.ContentType) domain.ContentTyp } } -func convertDomainContentTypeToProto(domainType domain.ContentType) pb.ContentType { - switch domainType { - case domain.ContentTypeHTML: - return pb.ContentType_CONTENT_TYPE_HTML - case domain.ContentTypeText: - return pb.ContentType_CONTENT_TYPE_TEXT - default: - return pb.ContentType_CONTENT_TYPE_TEXT - } -} - func convertDomainToProtoType(domainType domain.NotificationType) pb.NotificationType { switch domainType { case domain.TypeEmail: @@ -365,7 +370,7 @@ func convertDomainToProtoNotification(notif *domain.Notification) *pb.Notificati Id: notif.ID, Type: convertDomainToProtoType(notif.Type), Account: notif.Account, - Priority: pb.Priority(notif.Priority), + Priority: pb.Priority(clampInt32(int(notif.Priority))), Status: convertDomainToProtoStatus(notif.Status), Subject: notif.Subject, Body: notif.Body, @@ -373,8 +378,8 @@ func convertDomainToProtoNotification(notif *domain.Notification) *pb.Notificati Recipients: notif.Recipients, Metadata: convertInterfaceMapToString(notif.Metadata), CreatedAt: timestamppb.New(notif.CreatedAt), - RetryCount: int32(notif.RetryCount), - MaxRetries: int32(notif.MaxRetries), + RetryCount: clampInt32(notif.RetryCount), + MaxRetries: clampInt32(notif.MaxRetries), LastError: notif.LastError, } diff --git a/api/rest/cors_test.go b/api/rest/cors_test.go index de10ef3..7e88555 100644 --- a/api/rest/cors_test.go +++ b/api/rest/cors_test.go @@ -17,9 +17,9 @@ func TestCORSMiddleware_AllowedOrigin(t *testing.T) { } middleware := newCORSMiddleware(config) - handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) - w.Write([]byte("OK")) + _, _ = w.Write([]byte("OK")) })) tests := []struct { @@ -94,9 +94,9 @@ func TestCORSMiddleware_BlockedOrigin(t *testing.T) { } middleware := newCORSMiddleware(config) - handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) - w.Write([]byte("OK")) + _, _ = w.Write([]byte("OK")) })) tests := []struct { @@ -158,7 +158,7 @@ func TestCORSMiddleware_PreflightRequest(t *testing.T) { } middleware := newCORSMiddleware(config) - handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler := middleware(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { t.Error("Handler should not be called for OPTIONS request") })) @@ -217,7 +217,7 @@ func TestCORSMiddleware_Credentials(t *testing.T) { } middleware := newCORSMiddleware(config) - handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) @@ -243,7 +243,7 @@ func TestCORSMiddleware_NoWildcard(t *testing.T) { } middleware := newCORSMiddleware(config) - handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) @@ -273,7 +273,7 @@ func TestCORSMiddleware_EmptyConfig(t *testing.T) { } middleware := newCORSMiddleware(config) - handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) @@ -355,7 +355,7 @@ func TestCORSMiddleware_MaxAge(t *testing.T) { } middleware := newCORSMiddleware(config) - handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handler := middleware(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) diff --git a/api/rest/handlers.go b/api/rest/handlers.go index 2c00ad0..339ebbb 100644 --- a/api/rest/handlers.go +++ b/api/rest/handlers.go @@ -226,7 +226,7 @@ func (h *Handler) GetNotifiers(w http.ResponseWriter, r *http.Request) { } // HealthCheck handles GET /health -func (h *Handler) HealthCheck(w http.ResponseWriter, r *http.Request) { +func (h *Handler) HealthCheck(w http.ResponseWriter, _ *http.Request) { respondJSON(w, http.StatusOK, map[string]interface{}{ "status": "healthy", "service": "notifier", diff --git a/api/rest/keys.go b/api/rest/keys.go index d116699..701f47c 100644 --- a/api/rest/keys.go +++ b/api/rest/keys.go @@ -306,7 +306,7 @@ func (h *KeyManagementHandler) GetAuditLog(w http.ResponseWriter, r *http.Reques // Helper methods // hasRole checks if the auth context has a specific role -func (h *KeyManagementHandler) hasRole(authCtx *auth.AuthContext, role string) bool { +func (h *KeyManagementHandler) hasRole(authCtx *auth.Context, role string) bool { for _, r := range authCtx.Roles { if r == role { return true @@ -319,16 +319,24 @@ func (h *KeyManagementHandler) hasRole(authCtx *auth.AuthContext, role string) b func (h *KeyManagementHandler) respondJSON(w http.ResponseWriter, statusCode int, data interface{}) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(statusCode) - json.NewEncoder(w).Encode(data) + // Headers are already written at this point, so there's nothing left to + // do but log an encode failure. + if err := json.NewEncoder(w).Encode(data); err != nil { + h.logger.Errorf("Failed to encode JSON response: %v", err) + } } // respondError writes an error JSON response -func (h *KeyManagementHandler) respondError(w http.ResponseWriter, statusCode int, error string, message string) { +func (h *KeyManagementHandler) respondError(w http.ResponseWriter, statusCode int, errMsg string, message string) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(statusCode) resp := ErrorResponse{ - Error: error, + Error: errMsg, Message: message, } - json.NewEncoder(w).Encode(resp) + // Headers are already written at this point, so there's nothing left to + // do but log an encode failure. + if err := json.NewEncoder(w).Encode(resp); err != nil { + h.logger.Errorf("Failed to encode JSON error response: %v", err) + } } diff --git a/api/rest/router.go b/api/rest/router.go index 753d2ab..46f892b 100644 --- a/api/rest/router.go +++ b/api/rest/router.go @@ -1,3 +1,6 @@ +// Package rest implements the HTTP/JSON transport for the notifier +// service: request routing, handlers, authentication and CORS middleware, +// and API key management endpoints. package rest import ( @@ -138,9 +141,11 @@ func NewRouterWithOptions(opts RouterOptions) *mux.Router { // LivenessHandler returns a minimal liveness handler for dedicated health // listeners (the REST router serves the same signal at /health). func LivenessHandler() http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]interface{}{ + // Headers are already written, so an encode error can only be + // dropped on the floor here. + _ = json.NewEncoder(w).Encode(map[string]interface{}{ "status": "healthy", "service": "notifier", "time": time.Now().UTC(), @@ -173,7 +178,9 @@ func readinessHandler(checks map[string]ReadinessCheck) http.HandlerFunc { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) ready := status == http.StatusOK - json.NewEncoder(w).Encode(map[string]interface{}{ + // Headers are already written, so an encode error can only be + // dropped on the floor here. + _ = json.NewEncoder(w).Encode(map[string]interface{}{ "ready": ready, "components": components, }) diff --git a/cmd/client/main.go b/cmd/client/main.go index 6fe5f3b..7a4eabc 100644 --- a/cmd/client/main.go +++ b/cmd/client/main.go @@ -1,3 +1,5 @@ +// Command client is a CLI for sending and managing notifications through +// the notifier service's REST API. package main import ( @@ -107,7 +109,7 @@ Options: account := fs.String("account", "", "") recipients := fs.String("recipients", "", "") - fs.Parse(args) + _ = fs.Parse(args) // flag.ExitOnError means Parse never returns a non-nil error if *notifType == "" || *body == "" { fmt.Fprintf(os.Stderr, "Error: --type and --body are required\n") @@ -118,7 +120,7 @@ Options: ctx, cancel := context.WithTimeout(context.Background(), *timeout) defer cancel() - cfg := client.ClientConfig{ + cfg := client.Config{ BaseURL: *baseURL, APIKey: *apiKey, Timeout: *timeout, @@ -174,7 +176,7 @@ Options: timeout := fs.Duration("timeout", 30*time.Second, "") id := fs.String("id", "", "") - fs.Parse(args) + _ = fs.Parse(args) // flag.ExitOnError means Parse never returns a non-nil error if *id == "" { fmt.Fprintf(os.Stderr, "Error: --id is required\n") @@ -185,7 +187,7 @@ Options: ctx, cancel := context.WithTimeout(context.Background(), *timeout) defer cancel() - cfg := client.ClientConfig{ + cfg := client.Config{ BaseURL: *baseURL, APIKey: *apiKey, Timeout: *timeout, @@ -231,12 +233,12 @@ Options: limit := fs.Int("limit", 10, "") offset := fs.Int("offset", 0, "") - fs.Parse(args) + _ = fs.Parse(args) // flag.ExitOnError means Parse never returns a non-nil error ctx, cancel := context.WithTimeout(context.Background(), *timeout) defer cancel() - cfg := client.ClientConfig{ + cfg := client.Config{ BaseURL: *baseURL, APIKey: *apiKey, Timeout: *timeout, @@ -290,12 +292,12 @@ Options: apiKey := fs.String("key", "", "") timeout := fs.Duration("timeout", 30*time.Second, "") - fs.Parse(args) + _ = fs.Parse(args) // flag.ExitOnError means Parse never returns a non-nil error ctx, cancel := context.WithTimeout(context.Background(), *timeout) defer cancel() - cfg := client.ClientConfig{ + cfg := client.Config{ BaseURL: *baseURL, APIKey: *apiKey, Timeout: *timeout, @@ -333,12 +335,12 @@ Options: apiKey := fs.String("key", "", "") timeout := fs.Duration("timeout", 30*time.Second, "") - fs.Parse(args) + _ = fs.Parse(args) // flag.ExitOnError means Parse never returns a non-nil error ctx, cancel := context.WithTimeout(context.Background(), *timeout) defer cancel() - cfg := client.ClientConfig{ + cfg := client.Config{ BaseURL: *baseURL, APIKey: *apiKey, Timeout: *timeout, @@ -374,12 +376,12 @@ Options: baseURL := fs.String("url", "http://localhost:8080", "") timeout := fs.Duration("timeout", 30*time.Second, "") - fs.Parse(args) + _ = fs.Parse(args) // flag.ExitOnError means Parse never returns a non-nil error ctx, cancel := context.WithTimeout(context.Background(), *timeout) defer cancel() - cfg := client.ClientConfig{ + cfg := client.Config{ BaseURL: *baseURL, Timeout: *timeout, TLSInsecure: false, @@ -396,8 +398,7 @@ Options: if healthy { fmt.Println("Service is healthy") os.Exit(0) - } else { - fmt.Println("Service is unhealthy") - os.Exit(1) } + fmt.Println("Service is unhealthy") + os.Exit(1) } diff --git a/cmd/server/main.go b/cmd/server/main.go index 4c1731d..e12f2c9 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -1,3 +1,5 @@ +// Command server runs the notifier service, exposing its REST and gRPC +// APIs and wiring up configuration, queueing, auth, and metrics. package main import ( @@ -364,7 +366,7 @@ func registerNotifiers(cfg *config.Config, factory *notifier.Factory, logger *lo } } -func startGRPCServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config, svc domain.NotificationService, logger *logging.Logger, authStore *auth.APIKeyStore) *grpc.Server { +func startGRPCServer(_ context.Context, wg *sync.WaitGroup, cfg *config.Config, svc domain.NotificationService, logger *logging.Logger, authStore *auth.APIKeyStore) *grpc.Server { addr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.GRPCPort) lis, err := net.Listen("tcp", addr) @@ -425,7 +427,7 @@ func startGRPCServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config return grpcServer } -func startRESTServer(ctx context.Context, wg *sync.WaitGroup, cfg *config.Config, svc domain.NotificationService, logger *logging.Logger, authStore *auth.APIKeyStore, hybridKeyStore *auth.HybridKeyStore, readiness map[string]rest.ReadinessCheck, collector *metrics.Collector) *http.Server { +func startRESTServer(_ context.Context, wg *sync.WaitGroup, cfg *config.Config, svc domain.NotificationService, logger *logging.Logger, authStore *auth.APIKeyStore, hybridKeyStore *auth.HybridKeyStore, readiness map[string]rest.ReadinessCheck, collector *metrics.Collector) *http.Server { opts := rest.RouterOptions{ Service: svc, Logger: logger, diff --git a/internal/auth/auth.go b/internal/auth/auth.go index 07b2e34..cd368a2 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -1,3 +1,7 @@ +// Package auth provides API key authentication and authorization for the +// notifier service, including key storage backends (in-memory, database, +// and a hybrid cache-plus-database store) and RBAC-style notifier +// authorization. package auth import ( @@ -54,8 +58,8 @@ type RateLimiter struct { mu sync.Mutex } -// AuthContext holds auth information attached to request context -type AuthContext struct { +// Context holds auth information attached to request context +type Context struct { APIKey *APIKey ClientID string Roles []string @@ -283,12 +287,12 @@ func (s *APIKeyStore) ListKeys(clientID string) []*APIKey { type authContextKey struct{} // ContextWithAuth adds auth context to a request context -func ContextWithAuth(ctx context.Context, auth *AuthContext) context.Context { +func ContextWithAuth(ctx context.Context, auth *Context) context.Context { return context.WithValue(ctx, authContextKey{}, auth) } // GetAuthContext retrieves auth context from a request context -func GetAuthContext(ctx context.Context) (*AuthContext, bool) { - auth, ok := ctx.Value(authContextKey{}).(*AuthContext) +func GetAuthContext(ctx context.Context) (*Context, bool) { + auth, ok := ctx.Value(authContextKey{}).(*Context) return auth, ok } diff --git a/internal/auth/authz.go b/internal/auth/authz.go index f6ebb71..5020f31 100644 --- a/internal/auth/authz.go +++ b/internal/auth/authz.go @@ -26,7 +26,7 @@ func (a *NotifierAuthz) RegisterRule(notificationType domain.NotificationType, a } // IsAuthorized checks if an auth context is authorized to use a specific notifier -func (a *NotifierAuthz) IsAuthorized(auth *AuthContext, notificationType domain.NotificationType, account string) bool { +func (a *NotifierAuthz) IsAuthorized(auth *Context, notificationType domain.NotificationType, account string) bool { if auth == nil || len(auth.Roles) == 0 { return false } diff --git a/internal/auth/bootstrap.go b/internal/auth/bootstrap.go index a0c0d73..1bbb989 100644 --- a/internal/auth/bootstrap.go +++ b/internal/auth/bootstrap.go @@ -183,7 +183,7 @@ func BootstrapAdminKey(ctx context.Context, keyStore *HybridKeyStore, cfg *Boots // LoadBootstrapKeyFromEnv checks if a bootstrap key was provided via environment variable // This allows injecting a pre-generated key via CI/CD -func LoadBootstrapKeyFromEnv(ctx context.Context, keyStore *HybridKeyStore, logger *logging.Logger) error { +func LoadBootstrapKeyFromEnv(_ context.Context, _ *HybridKeyStore, logger *logging.Logger) error { bootstrapKey := os.Getenv("NOTIFIER_BOOTSTRAP_ADMIN_KEY") if bootstrapKey == "" { return nil // Not set, skip diff --git a/internal/auth/grpc_middleware.go b/internal/auth/grpc_middleware.go index e6baefa..d989b6e 100644 --- a/internal/auth/grpc_middleware.go +++ b/internal/auth/grpc_middleware.go @@ -55,7 +55,7 @@ func (m *GRPCAuthMiddleware) UnaryInterceptor() grpc.UnaryServerInterceptor { } // Create auth context and attach to request - authCtx := &AuthContext{ + authCtx := &Context{ APIKey: key, ClientID: key.ClientID, Roles: key.Roles, @@ -99,7 +99,7 @@ func (m *GRPCAuthMiddleware) StreamInterceptor() grpc.StreamServerInterceptor { } // Create auth context and attach to request - authCtx := &AuthContext{ + authCtx := &Context{ APIKey: key, ClientID: key.ClientID, Roles: key.Roles, diff --git a/internal/auth/keystore_db.go b/internal/auth/keystore_db.go index 177adc5..0a22af0 100644 --- a/internal/auth/keystore_db.go +++ b/internal/auth/keystore_db.go @@ -128,7 +128,7 @@ func (ks *KeyStoreDB) migrateLegacyPlaintextKeys() error { if err != nil { return fmt.Errorf("failed to read legacy keys: %w", err) } - defer rows.Close() + defer func() { _ = rows.Close() }() type legacyRow struct { id int @@ -265,7 +265,7 @@ func (ks *KeyStoreDB) ListKeys(ctx context.Context, clientID string) ([]*APIKey, if err != nil { return nil, fmt.Errorf("failed to list keys: %w", err) } - defer rows.Close() + defer func() { _ = rows.Close() }() var keys []*APIKey for rows.Next() { @@ -318,7 +318,7 @@ func (ks *KeyStoreDB) LoadAllKeys(ctx context.Context) ([]*APIKey, error) { if err != nil { return nil, fmt.Errorf("failed to load keys: %w", err) } - defer rows.Close() + defer func() { _ = rows.Close() }() var keys []*APIKey for rows.Next() { @@ -427,7 +427,7 @@ func (ks *KeyStoreDB) auditLogQuery(ctx context.Context, query string, ident str if err != nil { return nil, fmt.Errorf("failed to get audit log: %w", err) } - defer rows.Close() + defer func() { _ = rows.Close() }() var logs []map[string]interface{} for rows.Next() { diff --git a/internal/auth/keystore_test.go b/internal/auth/keystore_test.go index 3211f47..a36a136 100644 --- a/internal/auth/keystore_test.go +++ b/internal/auth/keystore_test.go @@ -68,7 +68,7 @@ func (f *fakeKeyDB) DeactivateKeyByHash(_ context.Context, keyHash string, _ str return nil } -func (f *fakeKeyDB) UpdateLastUsed(_ context.Context, keyHash string) error { return nil } +func (f *fakeKeyDB) UpdateLastUsed(_ context.Context, _ string) error { return nil } func (f *fakeKeyDB) LoadAllKeys(_ context.Context) ([]*APIKey, error) { var keys []*APIKey diff --git a/internal/auth/rest_middleware.go b/internal/auth/rest_middleware.go index 60153d9..a79ac22 100644 --- a/internal/auth/rest_middleware.go +++ b/internal/auth/rest_middleware.go @@ -55,7 +55,7 @@ func (m *RESTAuthMiddleware) Middleware(next http.Handler) http.Handler { } // Create auth context and attach to request - authCtx := &AuthContext{ + authCtx := &Context{ APIKey: key, ClientID: key.ClientID, Roles: key.Roles, diff --git a/internal/config/config.go b/internal/config/config.go index 09e4c9b..f8e7403 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,3 +1,6 @@ +// Package config loads and validates the notifier service's configuration +// (notifiers, queue, auth, retention, and related settings) from files, +// environment variables, and defaults via viper. package config import ( @@ -281,7 +284,7 @@ func (c *Config) Validate() error { } if c.Queue.Type == "kafka" && c.Queue.Kafka == nil { - return fmt.Errorf("Kafka queue type selected but no Kafka configuration provided") + return fmt.Errorf("kafka queue type selected but no kafka configuration provided") } // Validate at least one notifier is configured diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 4422daa..e546387 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -10,7 +10,7 @@ func TestSanitizeDatabaseURL(t *testing.T) { input string expected string }{ - { + { //nolint:gosec // test fixture URL, not a real credential name: "PostgreSQL with password", input: "postgresql://user:password@localhost:5432/dbname", expected: "postgresql://user:***REDACTED***@localhost:5432/dbname", @@ -20,7 +20,7 @@ func TestSanitizeDatabaseURL(t *testing.T) { input: "postgresql://user@localhost:5432/dbname", expected: "postgresql://user@localhost:5432/dbname", }, - { + { //nolint:gosec // test fixture URL, not a real credential name: "MySQL with special characters in password", input: "mysql://root:SuperSecret123!@db.example.com:3306/mydb", expected: "mysql://root:***REDACTED***@db.example.com:3306/mydb", @@ -40,12 +40,12 @@ func TestSanitizeDatabaseURL(t *testing.T) { input: "postgresql://localhost:5432/dbname", expected: "postgresql://localhost:5432/dbname", }, - { + { //nolint:gosec // test fixture URL, not a real credential name: "PostgreSQL with password containing colons", input: "postgresql://user:pass:word@localhost:5432/dbname", expected: "postgresql://user:***REDACTED***@localhost:5432/dbname", }, - { + { //nolint:gosec // test fixture URL, not a real credential name: "PostgreSQL with complex hostname and port", input: "postgresql://admin:p@ssw0rd!@db-prod.example.com:5432/production", expected: "postgresql://admin:***REDACTED***@db-prod.example.com:5432/production", diff --git a/internal/domain/notification.go b/internal/domain/notification.go index f397880..13736ba 100644 --- a/internal/domain/notification.go +++ b/internal/domain/notification.go @@ -1,3 +1,6 @@ +// Package domain contains the core types shared across the notifier +// service - notifications, queueing primitives, and the notifier +// interfaces that provider implementations satisfy. package domain import ( @@ -15,6 +18,7 @@ var ( // Priority defines the urgency level of a notification type Priority int +// Priority levels, in increasing order of urgency. const ( PriorityLow Priority = iota PriorityNormal @@ -25,6 +29,7 @@ const ( // NotificationType defines the channel through which to send the notification type NotificationType string +// Supported notification channels. const ( TypeEmail NotificationType = "email" TypeSlack NotificationType = "slack" @@ -35,6 +40,7 @@ const ( // ContentType defines the format of the notification body type ContentType string +// Supported body content types. const ( ContentTypeText ContentType = "text" ContentTypeHTML ContentType = "html" @@ -43,6 +49,7 @@ const ( // NotificationStatus represents the current state of a notification type NotificationStatus string +// Notification lifecycle states. const ( StatusPending NotificationStatus = "pending" StatusQueued NotificationStatus = "queued" diff --git a/internal/logging/logger.go b/internal/logging/logger.go index a07aee4..5f8481a 100644 --- a/internal/logging/logger.go +++ b/internal/logging/logger.go @@ -1,3 +1,6 @@ +// Package logging provides structured logging backed by log/slog, with UTC +// RFC3339 timestamps and a level-gated API compatible with the previous +// *log.Logger-based implementation. package logging import ( @@ -19,6 +22,8 @@ type Logger struct { // LogLevel represents the logging level type LogLevel int +// Logging levels, in increasing order of severity. DebugLevel is the most +// verbose and ErrorLevel the least. const ( DebugLevel LogLevel = iota InfoLevel @@ -102,7 +107,7 @@ func NewFromOptions(levelStr string, format string, outputPath string) (*Logger, case "stderr": output = os.Stderr default: - file, err := os.OpenFile(outputPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + file, err := os.OpenFile(outputPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) //nolint:gosec // outputPath is operator-configured (logging.output), not user-controlled input if err != nil { return nil, err } diff --git a/internal/logging/logger_test.go b/internal/logging/logger_test.go index 52b0193..6dccc93 100644 --- a/internal/logging/logger_test.go +++ b/internal/logging/logger_test.go @@ -12,16 +12,16 @@ import ( func TestNew_TextFormat(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "text.log") - file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("unexpected error opening file: %v", err) } - defer file.Close() + defer func() { _ = file.Close() }() logger := New(InfoLevel, file) logger.Info("hello world") - data, err := os.ReadFile(path) + data, err := os.ReadFile(path) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read log file: %v", err) } @@ -49,7 +49,7 @@ func TestNewFromConfig_JSONOutput(t *testing.T) { logger.Info("structured message") - data, err := os.ReadFile(path) + data, err := os.ReadFile(path) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read log file: %v", err) } @@ -77,17 +77,17 @@ func TestNewFromConfig_JSONOutput(t *testing.T) { func TestLevelFiltering_DebugSuppressedAtInfo(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "level.log") - file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("unexpected error opening file: %v", err) } - defer file.Close() + defer func() { _ = file.Close() }() logger := New(InfoLevel, file) logger.Debug("should not appear") logger.Info("should appear") - data, err := os.ReadFile(path) + data, err := os.ReadFile(path) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read log file: %v", err) } @@ -112,7 +112,7 @@ func TestNewFromOptions_FileOutput(t *testing.T) { logger.Info("file message") - data, err := os.ReadFile(path) + data, err := os.ReadFile(path) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read log file: %v", err) } @@ -136,7 +136,7 @@ func TestNewFromOptions_FormatSelection(t *testing.T) { } jsonLogger.Info("json message") - jsonData, err := os.ReadFile(jsonPath) + jsonData, err := os.ReadFile(jsonPath) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read json log file: %v", err) } @@ -151,7 +151,7 @@ func TestNewFromOptions_FormatSelection(t *testing.T) { } textLogger.Info("text message") - textData, err := os.ReadFile(textPath) + textData, err := os.ReadFile(textPath) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read text log file: %v", err) } @@ -170,7 +170,7 @@ func TestNewFromOptions_FormatSelection(t *testing.T) { } defaultLogger.Info("default message") - defaultData, err := os.ReadFile(defaultPath) + defaultData, err := os.ReadFile(defaultPath) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read default log file: %v", err) } @@ -186,7 +186,7 @@ func TestNewFromOptions_FormatSelection(t *testing.T) { } configLogger.Info("config message") - configData, err := os.ReadFile(configPath) + configData, err := os.ReadFile(configPath) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read config log file: %v", err) } @@ -198,11 +198,11 @@ func TestNewFromOptions_FormatSelection(t *testing.T) { func TestSlogAccessor(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "slog.log") - file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("unexpected error opening file: %v", err) } - defer file.Close() + defer func() { _ = file.Close() }() logger := New(InfoLevel, file) if logger.Slog() == nil { @@ -211,7 +211,7 @@ func TestSlogAccessor(t *testing.T) { logger.Slog().Info("via slog accessor") - data, err := os.ReadFile(path) + data, err := os.ReadFile(path) //nolint:gosec // path is from t.TempDir() if err != nil { t.Fatalf("failed to read log file: %v", err) } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index a1d18d4..1dd85ca 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -15,6 +15,7 @@ import ( "github.com/igodwin/notifier/internal/domain" "github.com/igodwin/notifier/internal/logging" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" "github.com/prometheus/client_golang/prometheus/promhttp" ) @@ -72,8 +73,8 @@ func NewCollector(service domain.NotificationService, queue domain.Queue, logger c.queueDepth, c.httpRequests, c.httpDuration, - prometheus.NewGoCollector(), - prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}), + collectors.NewGoCollector(), + collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), ) return c diff --git a/internal/notifier/notifier.go b/internal/notifier/notifier.go index be681fc..4e8e893 100644 --- a/internal/notifier/notifier.go +++ b/internal/notifier/notifier.go @@ -1,3 +1,6 @@ +// Package notifier defines the notifier provider interfaces and a factory +// for constructing and looking up configured notifier instances by type +// and account. package notifier import ( diff --git a/internal/notifier/ntfy.go b/internal/notifier/ntfy.go index 226cdbc..f3c505a 100644 --- a/internal/notifier/ntfy.go +++ b/internal/notifier/ntfy.go @@ -127,7 +127,7 @@ func validateCACertPath(caCertPath string) error { } // Try to read and parse the certificate - certData, err := os.ReadFile(caCertPath) + certData, err := os.ReadFile(caCertPath) //nolint:gosec // caCertPath is operator-configured (ntfy CA cert path), not user-controlled input if err != nil { return fmt.Errorf("failed to read CA certificate file: %w", err) } @@ -270,8 +270,8 @@ func (n *NtfyNotifier) Send(ctx context.Context, notification *domain.Notificati if body, ok := actionMap["body"].(string); ok { ntfyAct.Body = body } - if clear, ok := actionMap["clear"].(bool); ok { - ntfyAct.Clear = clear + if clearAction, ok := actionMap["clear"].(bool); ok { + ntfyAct.Clear = clearAction } req.Actions = append(req.Actions, ntfyAct) } @@ -302,7 +302,7 @@ func (n *NtfyNotifier) Send(ctx context.Context, notification *domain.Notificati // sendToTopic sends a notification to a specific ntfy topic func (n *NtfyNotifier) sendToTopic(ctx context.Context, req *ntfyRequest) error { - url := fmt.Sprintf("%s", n.config.ServerURL) + url := n.config.ServerURL jsonData, err := json.Marshal(req) if err != nil { @@ -327,7 +327,7 @@ func (n *NtfyNotifier) sendToTopic(ctx context.Context, req *ntfyRequest) error if err != nil { return fmt.Errorf("failed to send ntfy notification: %w", err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode < 200 || resp.StatusCode >= 300 { return fmt.Errorf("ntfy server returned status: %d", resp.StatusCode) diff --git a/internal/notifier/ntfy_tls_test.go b/internal/notifier/ntfy_tls_test.go index b97ac89..37204a4 100644 --- a/internal/notifier/ntfy_tls_test.go +++ b/internal/notifier/ntfy_tls_test.go @@ -48,7 +48,7 @@ func TestNewNtfyNotifierWithDefaultCA(t *testing.T) { func TestNewNtfyNotifierWithCustomCA(t *testing.T) { // Create a temporary CA certificate file certPath := createTempCACert(t) - defer os.Remove(certPath) + defer func() { _ = os.Remove(certPath) }() config := &NtfyConfig{ ServerURL: "https://self-signed.example.com", @@ -95,13 +95,13 @@ func TestValidateCACertPathInvalidFormat(t *testing.T) { if err != nil { t.Fatalf("Failed to create temp file: %v", err) } - defer os.Remove(tmpFile.Name()) + defer func() { _ = os.Remove(tmpFile.Name()) }() // Write invalid content (not PEM format) if _, err := tmpFile.WriteString("This is not a valid certificate"); err != nil { t.Fatalf("Failed to write to temp file: %v", err) } - tmpFile.Close() + _ = tmpFile.Close() config := &NtfyConfig{ ServerURL: "https://ntfy.sh", @@ -128,7 +128,7 @@ func TestValidateCACertPathIsDirectory(t *testing.T) { if err != nil { t.Fatalf("Failed to create temp directory: %v", err) } - defer os.RemoveAll(tmpDir) + defer func() { _ = os.RemoveAll(tmpDir) }() config := &NtfyConfig{ ServerURL: "https://ntfy.sh", @@ -210,7 +210,7 @@ func TestTLSConfigNeverSkipsVerification(t *testing.T) { func TestCustomCACertLoading(t *testing.T) { // Create a temporary CA certificate certPath := createTempCACert(t) - defer os.Remove(certPath) + defer func() { _ = os.Remove(certPath) }() config := &NtfyConfig{ ServerURL: "https://self-signed.example.com", @@ -257,8 +257,8 @@ func TestEmptyCertFileError(t *testing.T) { if err != nil { t.Fatalf("Failed to create temp file: %v", err) } - defer os.Remove(tmpFile.Name()) - tmpFile.Close() + defer func() { _ = os.Remove(tmpFile.Name()) }() + _ = tmpFile.Close() err = validateCACertPath(tmpFile.Name()) if err == nil { @@ -274,7 +274,7 @@ func createTempCACert(t *testing.T) string { if err != nil { t.Fatalf("Failed to create temp file: %v", err) } - defer tmpFile.Close() + defer func() { _ = tmpFile.Close() }() // Generate a self-signed certificate for testing certPEM := generateSelfSignedCert(t) diff --git a/internal/notifier/slack.go b/internal/notifier/slack.go index cb34f35..fa7de14 100644 --- a/internal/notifier/slack.go +++ b/internal/notifier/slack.go @@ -55,12 +55,12 @@ type slackTextBlock struct { // NewSlackNotifier creates a new Slack notifier func NewSlackNotifier(config *SlackConfig) (*SlackNotifier, error) { if config == nil { - return nil, fmt.Errorf("Slack config is required") + return nil, fmt.Errorf("slack config is required") } // Either webhook URL or token is required if config.WebhookURL == "" && config.Token == "" && len(config.Webhooks) == 0 { - return nil, fmt.Errorf("Slack webhook URL, token, or channel webhooks are required") + return nil, fmt.Errorf("slack webhook URL, token, or channel webhooks are required") } return &SlackNotifier{ @@ -201,10 +201,10 @@ func (s *SlackNotifier) sendToSlack(ctx context.Context, webhookURL string, msg if err != nil { return fmt.Errorf("failed to send Slack notification: %w", err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("Slack API returned status: %d", resp.StatusCode) + return fmt.Errorf("slack API returned status: %d", resp.StatusCode) } return nil diff --git a/internal/notifier/smtp.go b/internal/notifier/smtp.go index af62667..750e552 100644 --- a/internal/notifier/smtp.go +++ b/internal/notifier/smtp.go @@ -148,13 +148,13 @@ func sendMailImplicitTLS(addr, serverName string, auth smtp.Auth, from string, r if err != nil { return fmt.Errorf("failed to establish TLS connection to %s: %w", addr, err) } - defer conn.Close() + defer func() { _ = conn.Close() }() client, err := smtp.NewClient(conn, serverName) if err != nil { return fmt.Errorf("failed to create SMTP client: %w", err) } - defer client.Close() + defer func() { _ = client.Close() }() if auth != nil { if ok, _ := client.Extension("AUTH"); ok { @@ -224,23 +224,23 @@ func (s *SMTPNotifier) buildMessage(notification *domain.Notification) string { fromHeader = fmt.Sprintf("%s <%s>", encodeHeaderValue(s.config.FromName), s.config.From) } - builder.WriteString(fmt.Sprintf("From: %s\r\n", fromHeader)) + fmt.Fprintf(&builder, "From: %s\r\n", fromHeader) // Add To header (optional if only BCC is specified) if len(notification.Recipients) > 0 { - builder.WriteString(fmt.Sprintf("To: %s\r\n", strings.Join(notification.Recipients, ", "))) + fmt.Fprintf(&builder, "To: %s\r\n", strings.Join(notification.Recipients, ", ")) } // Add CC header (optional) if len(notification.CC) > 0 { - builder.WriteString(fmt.Sprintf("Cc: %s\r\n", strings.Join(notification.CC, ", "))) + fmt.Fprintf(&builder, "Cc: %s\r\n", strings.Join(notification.CC, ", ")) } // Note: BCC is intentionally NOT included in headers (that's the point of BCC!) // Subject is fully attacker-controlled, so it is always run through RFC 2047 encoding. // This neutralizes embedded CR/LF (and non-ASCII) instead of interpolating it raw. - builder.WriteString(fmt.Sprintf("Subject: %s\r\n", encodeHeaderValue(notification.Subject))) + fmt.Fprintf(&builder, "Subject: %s\r\n", encodeHeaderValue(notification.Subject)) builder.WriteString("MIME-Version: 1.0\r\n") switch { @@ -275,24 +275,24 @@ func isHTMLContent(notification *domain.Notification) bool { func (s *SMTPNotifier) buildMultipartMessage(builder *strings.Builder, plainText, htmlBody string) { boundary := generateBoundary() - builder.WriteString(fmt.Sprintf("Content-Type: multipart/alternative; boundary=\"%s\"\r\n", boundary)) + fmt.Fprintf(builder, "Content-Type: multipart/alternative; boundary=\"%s\"\r\n", boundary) builder.WriteString("\r\n") - builder.WriteString(fmt.Sprintf("--%s\r\n", boundary)) + fmt.Fprintf(builder, "--%s\r\n", boundary) builder.WriteString("Content-Type: text/plain; charset=UTF-8\r\n") builder.WriteString("Content-Transfer-Encoding: 7bit\r\n") builder.WriteString("\r\n") builder.WriteString(plainText) builder.WriteString("\r\n\r\n") - builder.WriteString(fmt.Sprintf("--%s\r\n", boundary)) + fmt.Fprintf(builder, "--%s\r\n", boundary) builder.WriteString("Content-Type: text/html; charset=UTF-8\r\n") builder.WriteString("Content-Transfer-Encoding: 7bit\r\n") builder.WriteString("\r\n") builder.WriteString(htmlBody) builder.WriteString("\r\n\r\n") - builder.WriteString(fmt.Sprintf("--%s--\r\n", boundary)) + fmt.Fprintf(builder, "--%s--\r\n", boundary) } // detectContentType auto-detects if the body is HTML diff --git a/internal/queue/local.go b/internal/queue/local.go index b3e2241..8d0da06 100644 --- a/internal/queue/local.go +++ b/internal/queue/local.go @@ -1,3 +1,6 @@ +// Package queue provides domain.Queue implementations used to buffer +// notifications between submission and delivery, including an in-memory +// LocalQueue with optional disk persistence. package queue import ( @@ -129,7 +132,7 @@ func (lq *LocalQueue) Dequeue(ctx context.Context) (*domain.QueueMessage, error) } // Ack acknowledges successful processing of a message -func (lq *LocalQueue) Ack(ctx context.Context, messageID string) error { +func (lq *LocalQueue) Ack(_ context.Context, messageID string) error { lq.mu.Lock() defer lq.mu.Unlock() @@ -188,14 +191,14 @@ func (lq *LocalQueue) Nack(ctx context.Context, messageID string, requeue bool) } // Size returns the current number of messages in the queue -func (lq *LocalQueue) Size(ctx context.Context) (int64, error) { +func (lq *LocalQueue) Size(_ context.Context) (int64, error) { lq.mu.RLock() defer lq.mu.RUnlock() return int64(len(lq.queue)), nil } // Purge removes all messages from the queue -func (lq *LocalQueue) Purge(ctx context.Context) error { +func (lq *LocalQueue) Purge(_ context.Context) error { lq.mu.Lock() defer lq.mu.Unlock() @@ -238,7 +241,7 @@ func (lq *LocalQueue) Close() error { } // HealthCheck verifies the queue is operational -func (lq *LocalQueue) HealthCheck(ctx context.Context) error { +func (lq *LocalQueue) HealthCheck(_ context.Context) error { lq.mu.RLock() defer lq.mu.RUnlock() @@ -260,7 +263,7 @@ func (lq *LocalQueue) persistToDiskSync() error { return fmt.Errorf("failed to marshal queue state: %w", err) } - if err := os.WriteFile(lq.persistPath, data, 0644); err != nil { + if err := os.WriteFile(lq.persistPath, data, 0600); err != nil { return fmt.Errorf("failed to write queue state: %w", err) } diff --git a/internal/service/service.go b/internal/service/service.go index cd046e7..a2afa7d 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -1,3 +1,7 @@ +// Package service implements the core notification service: queueing, +// worker-pool delivery with retry/backoff, in-memory notification tracking, +// retention cleanup, and multi-tenant access control on top of the +// domain and auth packages. package service import ( @@ -166,14 +170,12 @@ func (s *NotificationService) performCleanup() { // Track which notifications to delete var toDelete []string - var allNotifications []*domain.Notification - // First pass: identify expired notifications and collect all for sorting + // First pass: identify expired notifications for id, notification := range s.notifications { if notification.CreatedAt.Before(expiredBefore) { toDelete = append(toDelete, id) } - allNotifications = append(allNotifications, notification) } // Delete expired notifications @@ -218,7 +220,7 @@ func (s *NotificationService) performCleanup() { } // worker processes notifications from the queue -func (s *NotificationService) worker(ctx context.Context, id int) { +func (s *NotificationService) worker(ctx context.Context, _ int) { defer s.wg.Done() for { @@ -284,7 +286,9 @@ func (s *NotificationService) processNotification(ctx context.Context, msg *doma notification.ID, notification.Type, account, err) notification.Status = domain.StatusFailed notification.LastError = fmt.Sprintf("failed to create notifier: %v", err) - s.queue.Nack(ctx, msg.ID, false) + if nackErr := s.queue.Nack(ctx, msg.ID, false); nackErr != nil { + s.logger.Warnf("failed to nack message id=%s: %v", msg.ID, nackErr) + } s.updateNotification(notification) return } @@ -313,13 +317,17 @@ func (s *NotificationService) processNotification(ctx context.Context, msg *doma notification.Status = domain.StatusFailed s.logger.Errorf("Notification send failed permanently - id=%s, type=%s, account=%s, recipients=%v, attempts=%d, error=%s", notification.ID, notification.Type, account, notification.Recipients, notification.RetryCount, notification.LastError) - s.queue.Nack(ctx, msg.ID, false) // Don't requeue + if nackErr := s.queue.Nack(ctx, msg.ID, false); nackErr != nil { // Don't requeue + s.logger.Warnf("failed to nack message id=%s: %v", msg.ID, nackErr) + } } } else { notification.Status = domain.StatusSent now := time.Now() notification.SentAt = &now - s.queue.Ack(ctx, msg.ID) + if ackErr := s.queue.Ack(ctx, msg.ID); ackErr != nil { + s.logger.Warnf("failed to ack message id=%s: %v", msg.ID, ackErr) + } s.logger.Infof("Notification sent successfully - id=%s, type=%s, account=%s, recipients=%v", notification.ID, notification.Type, account, notification.Recipients) } @@ -335,7 +343,9 @@ func (s *NotificationService) processNotification(ctx context.Context, msg *doma func (s *NotificationService) scheduleRetry(ctx context.Context, msg *domain.QueueMessage, notification *domain.Notification) { delay := s.retryDelay(notification.RetryCount) if delay <= 0 { - s.queue.Nack(ctx, msg.ID, true) // Requeue immediately + if err := s.queue.Nack(ctx, msg.ID, true); err != nil { // Requeue immediately + s.logger.Warnf("failed to nack message id=%s: %v", msg.ID, err) + } return } @@ -348,7 +358,9 @@ func (s *NotificationService) scheduleRetry(ctx context.Context, msg *domain.Que select { case <-timer.C: - s.queue.Nack(ctx, msg.ID, true) // Requeue after backoff + if err := s.queue.Nack(ctx, msg.ID, true); err != nil { // Requeue after backoff + s.logger.Warnf("failed to nack message id=%s: %v", msg.ID, err) + } case <-ctx.Done(): s.abandonRetry(msg, notification) case <-s.stopChan: @@ -362,7 +374,9 @@ func (s *NotificationService) scheduleRetry(ctx context.Context, msg *domain.Que // retry is still pending, so the notification isn't left stuck in "retrying" // forever and no goroutine lingers past shutdown. func (s *NotificationService) abandonRetry(msg *domain.QueueMessage, notification *domain.Notification) { - s.queue.Nack(context.Background(), msg.ID, false) // Don't requeue + if err := s.queue.Nack(context.Background(), msg.ID, false); err != nil { // Don't requeue + s.logger.Warnf("failed to nack message id=%s: %v", msg.ID, err) + } notification.Status = domain.StatusFailed s.updateNotification(notification) } diff --git a/internal/service/service_race_test.go b/internal/service/service_race_test.go index f75bfb0..3524168 100644 --- a/internal/service/service_race_test.go +++ b/internal/service/service_race_test.go @@ -26,7 +26,7 @@ func TestConcurrentSendGetListNoRace(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() const numSenders = 8 const sendsPerSender = 25 diff --git a/internal/service/service_retention_test.go b/internal/service/service_retention_test.go index 22c22fb..0a42b10 100644 --- a/internal/service/service_retention_test.go +++ b/internal/service/service_retention_test.go @@ -58,7 +58,7 @@ func TestTTLBasedCleanup(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() // Create old notification (created 2 seconds ago) oldTime := time.Now().Add(-2 * time.Second) @@ -128,7 +128,7 @@ func TestMaxSizeEnforcement(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() // Create 10 notifications for i := 0; i < 10; i++ { @@ -179,7 +179,7 @@ func TestCleanupRemovesOldestFirst(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() // Create notifications with distinct times baseTime := time.Now() @@ -234,7 +234,7 @@ func TestCleanupDisabled(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() // Create old notification oldTime := time.Now().Add(-2 * time.Second) @@ -278,7 +278,7 @@ func TestCleanupConcurrency(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() // Create some initial notifications for i := 0; i < 10; i++ { @@ -404,10 +404,11 @@ func TestCleanupGracefulShutdown(t *testing.T) { t.Errorf("Stop failed: %v", stopErr) } - // Verify notifications are still intact after graceful shutdown - stats, err := svc.GetStats(context.Background()) - if err == nil && stats.TotalSent > 0 { - // This is expected - notifications should persist through shutdown + // Verify notifications are still intact after graceful shutdown - it's + // expected that notifications persist through shutdown, so there's + // nothing further to assert beyond GetStats succeeding. + if stats, err := svc.GetStats(context.Background()); err == nil { + t.Logf("stats after graceful shutdown: sent=%d", stats.TotalSent) } } @@ -432,7 +433,7 @@ func TestCleanupWithMixedNotificationStatuses(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() oldTime := time.Now().Add(-2 * time.Second) @@ -499,7 +500,7 @@ func TestCleanupPerformance(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() // Create 5000 old notifications startTime := time.Now() diff --git a/internal/service/service_retry_test.go b/internal/service/service_retry_test.go index 2595a06..9211cc6 100644 --- a/internal/service/service_retry_test.go +++ b/internal/service/service_retry_test.go @@ -20,7 +20,7 @@ type alwaysFailNotifier struct { calls []time.Time } -func (n *alwaysFailNotifier) Send(ctx context.Context, notification *domain.Notification) (*domain.NotificationResult, error) { +func (n *alwaysFailNotifier) Send(_ context.Context, notification *domain.Notification) (*domain.NotificationResult, error) { n.mu.Lock() n.calls = append(n.calls, time.Now()) n.mu.Unlock() @@ -35,7 +35,7 @@ func (n *alwaysFailNotifier) Send(ctx context.Context, notification *domain.Noti func (n *alwaysFailNotifier) Type() domain.NotificationType { return domain.TypeStdout } -func (n *alwaysFailNotifier) Validate(notification *domain.Notification) error { return nil } +func (n *alwaysFailNotifier) Validate(_ *domain.Notification) error { return nil } func (n *alwaysFailNotifier) Close() error { return nil } @@ -71,7 +71,7 @@ func createFailingTestService(t *testing.T, fail domain.Notifier) *NotificationS // waitForStatus polls GetNotification until it observes the notification in // the given status, or fails the test after timeout. -func waitForStatus(t *testing.T, svc *NotificationService, ctx context.Context, id string, status domain.NotificationStatus, timeout time.Duration) *domain.Notification { +func waitForStatus(ctx context.Context, t *testing.T, svc *NotificationService, id string, status domain.NotificationStatus, timeout time.Duration) *domain.Notification { t.Helper() deadline := time.Now().Add(timeout) @@ -102,7 +102,7 @@ func TestRetryBackoffExponentialDelaysRequeue(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() notification := &domain.Notification{ ID: "backoff-exponential-1", @@ -116,7 +116,7 @@ func TestRetryBackoffExponentialDelaysRequeue(t *testing.T) { t.Fatalf("Send failed: %v", err) } - waitForStatus(t, svc, ctx, notification.ID, domain.StatusFailed, 5*time.Second) + waitForStatus(ctx, t, svc, notification.ID, domain.StatusFailed, 5*time.Second) calls := fail.callTimes() if len(calls) != 3 { @@ -152,7 +152,7 @@ func TestRetryBackoffNoneIsImmediate(t *testing.T) { if err := svc.Start(ctx); err != nil { t.Fatalf("Failed to start service: %v", err) } - defer svc.Stop() + defer func() { _ = svc.Stop() }() notification := &domain.Notification{ ID: "backoff-none-1", @@ -167,7 +167,7 @@ func TestRetryBackoffNoneIsImmediate(t *testing.T) { t.Fatalf("Send failed: %v", err) } - waitForStatus(t, svc, ctx, notification.ID, domain.StatusFailed, 2*time.Second) + waitForStatus(ctx, t, svc, notification.ID, domain.StatusFailed, 2*time.Second) elapsed := time.Since(start) if elapsed > 1*time.Second { diff --git a/internal/service/service_tenant_test.go b/internal/service/service_tenant_test.go index 23140df..07ccb05 100644 --- a/internal/service/service_tenant_test.go +++ b/internal/service/service_tenant_test.go @@ -10,11 +10,11 @@ import ( "github.com/igodwin/notifier/internal/domain" ) -// ctxForClient builds a context carrying an auth.AuthContext for the given +// ctxForClient builds a context carrying an auth.Context for the given // client and roles, as REST/gRPC middleware would attach after authenticating // a request. func ctxForClient(clientID string, roles ...string) context.Context { - return auth.ContextWithAuth(context.Background(), &auth.AuthContext{ + return auth.ContextWithAuth(context.Background(), &auth.Context{ ClientID: clientID, Roles: roles, }) diff --git a/pkg/client/rest.go b/pkg/client/rest.go index 510cd4d..900c543 100644 --- a/pkg/client/rest.go +++ b/pkg/client/rest.go @@ -8,6 +8,8 @@ import ( "fmt" "io" "net/http" + "net/url" + "strconv" "time" ) @@ -22,7 +24,7 @@ type RESTClient struct { } // NewRESTClient creates a new REST client with the given config -func NewRESTClient(cfg ClientConfig) *RESTClient { +func NewRESTClient(cfg Config) *RESTClient { if cfg.Timeout == 0 { cfg.Timeout = 30 * time.Second } @@ -34,7 +36,7 @@ func NewRESTClient(cfg ClientConfig) *RESTClient { } tlsConfig := &tls.Config{ - InsecureSkipVerify: cfg.TLSInsecure, + InsecureSkipVerify: cfg.TLSInsecure, // #nosec G402 -- explicit user opt-in (TLSInsecure) for self-signed test endpoints } httpClient := &http.Client{ @@ -134,9 +136,33 @@ func (c *RESTClient) GetNotification(ctx context.Context, id string) (*Notificat return ¬if, nil } -// ListNotifications lists notifications with filters +// ListNotifications lists notifications with filters. Filter fields are +// encoded as query parameters matching the server's parseNotificationFilter +// (limit, offset, repeated type/status/recipient). func (c *RESTClient) ListNotifications(ctx context.Context, filter ListNotificationsRequest) (*ListNotificationsResponse, error) { - respBody, statusCode, err := c.doRequest(ctx, "GET", "/api/v1/notifications", nil) + query := url.Values{} + if filter.Limit > 0 { + query.Set("limit", strconv.Itoa(filter.Limit)) + } + if filter.Offset > 0 { + query.Set("offset", strconv.Itoa(filter.Offset)) + } + for _, t := range filter.Types { + query.Add("type", t) + } + for _, s := range filter.Statuses { + query.Add("status", string(s)) + } + for _, r := range filter.Recipients { + query.Add("recipient", r) + } + + path := "/api/v1/notifications" + if encoded := query.Encode(); encoded != "" { + path += "?" + encoded + } + + respBody, statusCode, err := c.doRequest(ctx, "GET", path, nil) if err != nil { return nil, err } @@ -238,7 +264,7 @@ func (c *RESTClient) HealthCheck(ctx context.Context) (bool, error) { if err != nil { return false, fmt.Errorf("health check failed: %w", err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() return resp.StatusCode == http.StatusOK, nil } @@ -281,7 +307,7 @@ func (c *RESTClient) doRequest(ctx context.Context, method, path string, body [] } respBody, err := io.ReadAll(resp.Body) - resp.Body.Close() + _ = resp.Body.Close() if err != nil { lastErr = fmt.Errorf("failed to read response: %w", err) diff --git a/pkg/client/types.go b/pkg/client/types.go index e94df85..1d72848 100644 --- a/pkg/client/types.go +++ b/pkg/client/types.go @@ -1,3 +1,5 @@ +// Package client provides a Go client library and types for interacting +// with the notifier service's REST API. package client import "time" @@ -24,6 +26,7 @@ type NotificationResponse struct { // NotificationStatus represents the status of a notification type NotificationStatus string +// Notification status values returned by the notifier service. const ( StatusPending NotificationStatus = "pending" StatusQueued NotificationStatus = "queued" @@ -89,8 +92,8 @@ type NotifiersResponse struct { Notifiers []NotifierInfo `json:"notifiers"` } -// ClientConfig contains configuration for the client -type ClientConfig struct { +// Config contains configuration for the client +type Config struct { BaseURL string // Base URL for REST API (e.g., "http://localhost:8080") APIKey string // Optional API key for authentication Timeout time.Duration // Request timeout (default: 30s) @@ -100,3 +103,7 @@ type ClientConfig struct { // NEVER set this to true in production. Use proper certificates or provide custom CA certificates instead. TLSInsecure bool } + +// ClientConfig is a backward-compatible alias for Config. +// Deprecated: use Config. +type ClientConfig = Config //nolint:revive // kept for API compatibility diff --git a/tests/e2e/critical_1_test.go b/tests/e2e/critical_1_test.go index 0270852..c9aab7a 100644 --- a/tests/e2e/critical_1_test.go +++ b/tests/e2e/critical_1_test.go @@ -94,7 +94,6 @@ func TestCRITICAL1_MaxSizeEnforcement(t *testing.T) { // Send more notifications than max_size notificationCount := 10 - notificationIDs := make([]string, 0, notificationCount) for i := 0; i < notificationCount; i++ { req := client.NotificationRequest{ @@ -104,11 +103,10 @@ func TestCRITICAL1_MaxSizeEnforcement(t *testing.T) { Recipients: []string{"test@example.com"}, } - resp, err := suite.Client.Send(ctx, req) + _, err := suite.Client.Send(ctx, req) if err != nil { t.Fatalf("Failed to send notification %d: %v", i, err) } - notificationIDs = append(notificationIDs, resp.NotificationID) } t.Logf("Sent %d notifications", notificationCount) @@ -328,7 +326,7 @@ func TestCRITICAL1_MemoryBounded(t *testing.T) { req := client.NotificationRequest{ Type: "stdout", Subject: fmt.Sprintf("Batch %d Notif %d", batch, i), - Body: fmt.Sprintf("Test data for notification"), + Body: "Test data for notification", Recipients: []string{"test@example.com"}, } @@ -382,7 +380,7 @@ func TestCRITICAL1_ServiceHealthy(t *testing.T) { req := client.NotificationRequest{ Type: "stdout", Subject: fmt.Sprintf("Health %d", i), - Body: fmt.Sprintf("Test"), + Body: "Test", Recipients: []string{"test@example.com"}, } diff --git a/tests/e2e/suite_test.go b/tests/e2e/suite_test.go index 36946c8..e514591 100644 --- a/tests/e2e/suite_test.go +++ b/tests/e2e/suite_test.go @@ -94,13 +94,17 @@ func SetupSuite(t *testing.T, retention ...string) *TestSuite { // Get container port host, err := container.Host(ctx) if err != nil { - container.Terminate(ctx) + if termErr := container.Terminate(ctx); termErr != nil { + t.Logf("Failed to terminate container during cleanup: %v", termErr) + } t.Fatalf("Failed to get container host: %v", err) } port, err := container.MappedPort(ctx, "8080") if err != nil { - container.Terminate(ctx) + if termErr := container.Terminate(ctx); termErr != nil { + t.Logf("Failed to terminate container during cleanup: %v", termErr) + } t.Fatalf("Failed to get container port: %v", err) } @@ -118,7 +122,9 @@ func SetupSuite(t *testing.T, retention ...string) *TestSuite { deadline := time.Now().Add(30 * time.Second) for { if time.Now().After(deadline) { - container.Terminate(ctx) + if termErr := container.Terminate(ctx); termErr != nil { + t.Logf("Failed to terminate container during cleanup: %v", termErr) + } t.Fatalf("Service failed to become ready") } @@ -141,7 +147,13 @@ func SetupSuite(t *testing.T, retention ...string) *TestSuite { // TeardownSuite stops and removes the container func (s *TestSuite) TeardownSuite(ctx context.Context) { if s.Container != nil { - s.Container.Terminate(ctx) + if err := s.Container.Terminate(ctx); err != nil { + if s.T != nil { + s.T.Logf("Failed to terminate container during cleanup: %v", err) + } else { + fmt.Printf("e2e: failed to terminate container during cleanup: %v\n", err) + } + } } } @@ -172,7 +184,7 @@ func (s *TestSuite) GetLogs(ctx context.Context) string { if err != nil { return fmt.Sprintf("error reading logs: %v", err) } - defer reader.Close() + defer func() { _ = reader.Close() }() logs, err := io.ReadAll(reader) if err != nil { @@ -181,31 +193,3 @@ func (s *TestSuite) GetLogs(ctx context.Context) string { return string(logs) } - -// buildTestImage builds the docker image for testing -func buildTestImage(t *testing.T) { - // Get the project root - wd, err := os.Getwd() - if err != nil { - t.Fatalf("Failed to get working directory: %v", err) - } - - // Find the project root by looking for go.mod - for { - if _, err := os.Stat(filepath.Join(wd, "go.mod")); err == nil { - break - } - parent := filepath.Dir(wd) - if parent == wd { - t.Fatalf("Could not find project root") - } - wd = parent - } - - // Check if Dockerfile exists - dockerfile := filepath.Join(wd, "Dockerfile") - if _, err := os.Stat(dockerfile); err != nil { - t.Logf("Warning: Dockerfile not found at %s, using generic build", dockerfile) - // The container will be built from the binary - } -}