// 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 ( "context" "fmt" "sync" "time" "github.com/igodwin/notifier/internal/auth" "github.com/igodwin/notifier/internal/config" "github.com/igodwin/notifier/internal/domain" "github.com/igodwin/notifier/internal/logging" ) // AccountResolver is an interface for resolving default accounts type AccountResolver interface { GetDefaultAccount(notifierType domain.NotificationType) string } const ( // defaultRetryBaseDelay is the base delay used for exponential retry // backoff: delay = defaultRetryBaseDelay * 2^(RetryCount-1). defaultRetryBaseDelay = time.Second // maxRetryBackoffDelay caps the exponential backoff delay so a // persistently failing notification doesn't wait arbitrarily long between // attempts. maxRetryBackoffDelay = 30 * time.Second // adminRole grants access to all tenants' notifications regardless of // ClientID. adminRole = "admin" ) // NotificationService implements the domain.NotificationService interface type NotificationService struct { factory domain.NotifierFactory queue domain.Queue accountResolver AccountResolver authz *auth.NotifierAuthz notifications map[string]*domain.Notification mu sync.RWMutex workerCount int stopChan chan struct{} runCancel context.CancelFunc wg sync.WaitGroup logger *logging.Logger retentionConfig config.NotificationRetentionConfig cleanupStopChan chan struct{} ttlDuration time.Duration checkFrequencyDuration time.Duration retryBackoff string retryBaseDelay time.Duration } // NewNotificationService creates a new notification service func NewNotificationService(factory domain.NotifierFactory, queue domain.Queue, workerCount int, accountResolver AccountResolver, authz *auth.NotifierAuthz, logger *logging.Logger) *NotificationService { if workerCount <= 0 { workerCount = 10 } return &NotificationService{ factory: factory, queue: queue, accountResolver: accountResolver, authz: authz, notifications: make(map[string]*domain.Notification), workerCount: workerCount, stopChan: make(chan struct{}), logger: logger, cleanupStopChan: make(chan struct{}), retryBaseDelay: defaultRetryBaseDelay, } } // WithRetryBackoff sets the retry backoff strategy used when requeueing a // notification after a retryable send failure. Recognized values are // "exponential" (delay = 1s * 2^(RetryCount-1), capped at 30s) and "none" // (requeue immediately). An empty string is treated as "exponential", matching // the documented default for queue.retry_backoff. func (s *NotificationService) WithRetryBackoff(mode string) { s.retryBackoff = mode } // WithRetentionConfig sets the notification retention configuration func (s *NotificationService) WithRetentionConfig(cfg config.NotificationRetentionConfig) error { s.retentionConfig = cfg // Parse TTL duration ttl, err := time.ParseDuration(cfg.TTL) if err != nil { return fmt.Errorf("invalid TTL duration: %w", err) } s.ttlDuration = ttl // Parse check frequency duration checkFreq, err := time.ParseDuration(cfg.CheckFrequency) if err != nil { return fmt.Errorf("invalid check frequency duration: %w", err) } s.checkFrequencyDuration = checkFreq return nil } // Start starts the worker pool and cleanup goroutine func (s *NotificationService) Start(ctx context.Context) error { // Derive a service-lifetime context so Stop() can interrupt workers // blocked in Dequeue instead of waiting out their poll timeout. runCtx, cancel := context.WithCancel(ctx) s.runCancel = cancel for i := 0; i < s.workerCount; i++ { s.wg.Add(1) go s.worker(runCtx, i) } // Start cleanup goroutine if retention is enabled if s.retentionConfig.Enabled && s.checkFrequencyDuration > 0 { s.wg.Add(1) go s.cleanupLoop(runCtx) } return nil } // Stop stops the service gracefully func (s *NotificationService) Stop() error { close(s.stopChan) close(s.cleanupStopChan) if s.runCancel != nil { s.runCancel() } s.wg.Wait() return s.queue.Close() } // cleanupLoop runs at regular intervals to clean up old or excessive notifications func (s *NotificationService) cleanupLoop(ctx context.Context) { defer s.wg.Done() ticker := time.NewTicker(s.checkFrequencyDuration) defer ticker.Stop() for { select { case <-s.cleanupStopChan: s.logger.Debugf("Cleanup loop stopped") return case <-ctx.Done(): s.logger.Debugf("Cleanup loop context cancelled") return case <-ticker.C: s.performCleanup() } } } // performCleanup removes expired notifications and enforces maximum size limit func (s *NotificationService) performCleanup() { s.mu.Lock() defer s.mu.Unlock() now := time.Now() expiredBefore := now.Add(-s.ttlDuration) // Track which notifications to delete var toDelete []string // First pass: identify expired notifications for id, notification := range s.notifications { if notification.CreatedAt.Before(expiredBefore) { toDelete = append(toDelete, id) } } // Delete expired notifications for _, id := range toDelete { delete(s.notifications, id) } expiredCount := len(toDelete) // Second pass: enforce max size limit by removing oldest notifications if s.retentionConfig.MaxSize > 0 && len(s.notifications) > s.retentionConfig.MaxSize { excessCount := len(s.notifications) - s.retentionConfig.MaxSize // Sort remaining notifications by creation time (oldest first) remaining := make([]*domain.Notification, 0, len(s.notifications)) for _, notification := range s.notifications { remaining = append(remaining, notification) } // Simple bubble sort to find oldest notifications (more efficient alternatives available) for i := 0; i < len(remaining)-1; i++ { for j := 0; j < len(remaining)-i-1; j++ { if remaining[j].CreatedAt.After(remaining[j+1].CreatedAt) { remaining[j], remaining[j+1] = remaining[j+1], remaining[j] } } } // Delete the oldest excessCount notifications for i := 0; i < excessCount && i < len(remaining); i++ { delete(s.notifications, remaining[i].ID) } } currentSize := len(s.notifications) // Log cleanup statistics if expiredCount > 0 || currentSize > s.retentionConfig.MaxSize { s.logger.Infof("Cleanup completed - expired=%d, current_size=%d, max_size=%d", expiredCount, currentSize, s.retentionConfig.MaxSize) } } // worker processes notifications from the queue func (s *NotificationService) worker(ctx context.Context, _ int) { defer s.wg.Done() for { select { case <-s.stopChan: return case <-ctx.Done(): return default: // Try to dequeue with timeout workerCtx, cancel := context.WithTimeout(ctx, 5*time.Second) msg, err := s.queue.Dequeue(workerCtx) cancel() if err != nil { if err == context.DeadlineExceeded { continue } time.Sleep(100 * time.Millisecond) continue } if msg == nil { time.Sleep(100 * time.Millisecond) continue } // Process the notification s.processNotification(ctx, msg) } } } // processNotification sends a notification and handles the result func (s *NotificationService) processNotification(ctx context.Context, msg *domain.QueueMessage) { // Work on our own clone of the queued notification. msg.Notification is // owned by the queue (which mutates its Status for its own bookkeeping); // cloning here means our mutations below never race with the queue's or // with clones already handed out by GetNotification/ListNotifications. notification := msg.Notification.Clone() // Derive retry progress from msg.Attempt (incremented by the queue on // every dequeue): the queue's copy never sees the counter on our clone, // so the message itself is the source of truth across requeues. notification.RetryCount = msg.Attempt - 1 if notification.RetryCount < 0 { notification.RetryCount = 0 } s.logger.Debugf("Processing notification - id=%s, type=%s, recipients=%d", notification.ID, notification.Type, len(notification.Recipients)) // Resolve account if not specified account := notification.Account if account == "" && s.accountResolver != nil { account = s.accountResolver.GetDefaultAccount(notification.Type) } // Get the appropriate notifier notifier, err := s.factory.Create(notification.Type, account) if err != nil { s.logger.Errorf("Failed to create notifier - id=%s, type=%s, account=%s, error=%v", notification.ID, notification.Type, account, err) notification.Status = domain.StatusFailed notification.LastError = fmt.Sprintf("failed to create notifier: %v", err) 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 } // Send the notification result, err := notifier.Send(ctx, notification) if err != nil || result == nil || !result.Success { notification.RetryCount++ if result != nil { notification.LastError = result.Error } if err != nil { notification.LastError = err.Error() } // Check if we should retry if notification.RetryCount < notification.MaxRetries { notification.Status = domain.StatusRetrying s.logger.Warnf("Notification send failed, will retry - id=%s, type=%s, account=%s, attempt=%d/%d, error=%s", notification.ID, notification.Type, account, notification.RetryCount, notification.MaxRetries, notification.LastError) // Hand the retry goroutine its own clone: the worker still // publishes this notification via updateNotification below, and // a shutdown-time abandonRetry must not write to the same object. s.scheduleRetry(ctx, msg, notification.Clone()) } else { 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) 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 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) } s.updateNotification(notification) } // scheduleRetry requeues a failed notification, optionally delaying the // requeue according to the configured backoff strategy. When a delay applies, // the wait happens on a goroutine tracked by the service WaitGroup so Stop() // blocks until it finishes - guaranteeing the queue is never closed while a // requeue for it is still pending. func (s *NotificationService) scheduleRetry(ctx context.Context, msg *domain.QueueMessage, notification *domain.Notification) { delay := s.retryDelay(notification.RetryCount) if delay <= 0 { 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 } s.wg.Add(1) go func() { defer s.wg.Done() timer := time.NewTimer(delay) defer timer.Stop() select { case <-timer.C: 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: s.abandonRetry(msg, notification) } }() } // abandonRetry marks a notification as permanently failed without requeueing // it. It is used when the service shuts down while a backoff delay for a // 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) { 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) } // retryDelay computes how long to wait before requeueing a notification that // failed on attempt retryCount, per the configured backoff strategy. "none" // means requeue immediately (delay of 0); anything else - including the // empty string, the documented default - uses exponential backoff: base 1s * // 2^(retryCount-1), capped at maxRetryBackoffDelay. func (s *NotificationService) retryDelay(retryCount int) time.Duration { if s.retryBackoff == "none" { return 0 } base := s.retryBaseDelay if base <= 0 { base = defaultRetryBaseDelay } if retryCount < 1 { retryCount = 1 } shift := retryCount - 1 if shift > 20 { // guard against overflow for pathological retry counts shift = 20 } delay := base * time.Duration(int64(1)< maxRetryBackoffDelay { delay = maxRetryBackoffDelay } return delay } // Send queues a notification for delivery func (s *NotificationService) Send(ctx context.Context, notification *domain.Notification) (*domain.NotificationResult, error) { // Enforce RBAC authorization if configured if err := s.checkAuthorization(ctx, notification); err != nil { return &domain.NotificationResult{ NotificationID: notification.ID, Success: false, Error: err.Error(), SentAt: time.Now(), }, err } // Stamp tenant ownership from the auth context, if present. Absent auth // context (auth disabled) leaves ClientID empty, preserving current // behavior. if authCtx, ok := auth.GetAuthContext(ctx); ok && authCtx != nil { notification.ClientID = authCtx.ClientID } // Mark queued and store BEFORE enqueueing: a worker can pick the message // up immediately, and a post-enqueue status write here would overwrite // the worker's Sent/Failed transition. notification.Status = domain.StatusQueued s.storeNotification(notification) // Enqueue a clone: the queue mutates Status on its copy for its own // bookkeeping and workers clone again on dequeue, so no notification // object is ever shared between the queue, the store, and callers. if err := s.queue.Enqueue(ctx, notification.Clone()); err != nil { notification.Status = domain.StatusFailed notification.LastError = fmt.Sprintf("failed to enqueue: %v", err) s.updateNotification(notification) return &domain.NotificationResult{ NotificationID: notification.ID, Success: false, Error: notification.LastError, SentAt: time.Now(), }, err } return &domain.NotificationResult{ NotificationID: notification.ID, Success: true, Message: "notification queued successfully", SentAt: time.Now(), }, nil } // SendBatch queues multiple notifications for delivery func (s *NotificationService) SendBatch(ctx context.Context, notifications []*domain.Notification) ([]*domain.NotificationResult, error) { results := make([]*domain.NotificationResult, 0, len(notifications)) // Enforce RBAC authorization for each notification for _, notification := range notifications { if err := s.checkAuthorization(ctx, notification); err != nil { return nil, fmt.Errorf("authorization denied for notification type=%s account=%s: %w", notification.Type, notification.Account, err) } } // Stamp tenant ownership from the auth context, if present. if authCtx, ok := auth.GetAuthContext(ctx); ok && authCtx != nil { for _, notification := range notifications { notification.ClientID = authCtx.ClientID } } // Mark queued and store BEFORE enqueueing (see Send), then hand the // queue clones so it never shares notification objects with the store // or callers. queued := make([]*domain.Notification, 0, len(notifications)) for _, notification := range notifications { notification.Status = domain.StatusQueued s.storeNotification(notification) queued = append(queued, notification.Clone()) } if err := s.queue.EnqueueBatch(ctx, queued); err != nil { return nil, fmt.Errorf("failed to enqueue batch: %w", err) } // Create results for _, notification := range notifications { results = append(results, &domain.NotificationResult{ NotificationID: notification.ID, Success: true, Message: "notification queued successfully", SentAt: time.Now(), }) } return results, nil } // GetNotification retrieves a notification by ID. If an auth context is // present and the caller lacks the admin role, a notification belonging to a // different tenant is reported as not found rather than leaking its // existence. func (s *NotificationService) GetNotification(ctx context.Context, id string) (*domain.Notification, error) { s.mu.RLock() notification, exists := s.notifications[id] s.mu.RUnlock() if !exists || !s.tenantCanAccess(ctx, notification) { return nil, fmt.Errorf("%w: %s", domain.ErrNotificationNotFound, id) } return notification.Clone(), nil } // ListNotifications retrieves notifications matching the filter, scoped to // the caller's tenant unless they have the admin role or no auth context is // present. func (s *NotificationService) ListNotifications(ctx context.Context, filter *domain.NotificationFilter) ([]*domain.Notification, error) { s.mu.RLock() defer s.mu.RUnlock() // Simple in-memory filtering var results []*domain.Notification for _, notification := range s.notifications { if !s.tenantCanAccess(ctx, notification) { continue } if s.matchesFilter(notification, filter) { results = append(results, notification.Clone()) } } // Apply limit and offset if filter.Offset > 0 && filter.Offset < len(results) { results = results[filter.Offset:] } if filter.Limit > 0 && filter.Limit < len(results) { results = results[:filter.Limit] } return results, nil } // CancelNotification cancels a pending notification. A notification belonging // to a different tenant is reported as not found, matching GetNotification. func (s *NotificationService) CancelNotification(ctx context.Context, id string) error { s.mu.Lock() defer s.mu.Unlock() notification, exists := s.notifications[id] if !exists || !s.tenantCanAccess(ctx, notification) { return fmt.Errorf("%w: %s", domain.ErrNotificationNotFound, id) } if notification.Status == domain.StatusSent { return domain.ErrNotificationAlreadySent } updated := notification.Clone() updated.Status = domain.StatusFailed updated.LastError = "cancelled by user" s.notifications[id] = updated return nil } // RetryNotification retries a failed notification. Cross-tenant access is // rejected the same way as GetNotification (via the same not-found error), // since RetryNotification is built on top of it. func (s *NotificationService) RetryNotification(ctx context.Context, id string) (*domain.NotificationResult, error) { notification, err := s.GetNotification(ctx, id) if err != nil { return nil, err } if notification.Status == domain.StatusSent { return &domain.NotificationResult{ NotificationID: id, Success: false, Error: "notification already sent", SentAt: time.Now(), }, domain.ErrNotificationAlreadySent } // Reset retry count and status notification.RetryCount = 0 notification.Status = domain.StatusPending // Re-enqueue return s.Send(ctx, notification) } // GetStats returns notification statistics, scoped to the caller's tenant // unless they have the admin role or no auth context is present. func (s *NotificationService) GetStats(ctx context.Context) (*domain.NotificationStats, error) { s.mu.RLock() defer s.mu.RUnlock() stats := &domain.NotificationStats{ ByType: make(map[string]int64), ByStatus: make(map[string]int64), } for _, notification := range s.notifications { if !s.tenantCanAccess(ctx, notification) { continue } switch notification.Status { case domain.StatusSent: stats.TotalSent++ case domain.StatusFailed: stats.TotalFailed++ case domain.StatusPending: stats.TotalPending++ case domain.StatusQueued: stats.TotalQueued++ } stats.ByType[string(notification.Type)]++ stats.ByStatus[string(notification.Status)]++ } return stats, nil } // GetNotifiers returns information about available notifiers, filtered by authorization if auth context is provided func (s *NotificationService) GetNotifiers(ctx context.Context) (*domain.NotifiersResponse, error) { // Extract auth context from request context if available authCtx, _ := auth.GetAuthContext(ctx) supportedTypes := s.factory.SupportedTypes() notifiers := make([]domain.NotifierInfo, 0, len(supportedTypes)) for _, notifType := range supportedTypes { accounts := s.factory.GetAccounts(notifType) // Filter accounts by authorization if auth context is available and authz is configured if authCtx != nil && s.authz != nil { authorizedAccounts := make([]string, 0, len(accounts)) for _, account := range accounts { if s.authz.IsAuthorized(authCtx, notifType, account) { authorizedAccounts = append(authorizedAccounts, account) } } accounts = authorizedAccounts } // Skip notifier type if no authorized accounts if len(accounts) == 0 && authCtx != nil { continue } defaultAccount := "" if s.accountResolver != nil { defaultAccount = s.accountResolver.GetDefaultAccount(notifType) } // If default account was filtered out, clear it if authCtx != nil && s.authz != nil && defaultAccount != "" { if !s.authz.IsAuthorized(authCtx, notifType, defaultAccount) { defaultAccount = "" // If available, use first authorized account as default if len(accounts) > 0 { defaultAccount = accounts[0] } } } notifiers = append(notifiers, domain.NotifierInfo{ Type: notifType, Accounts: accounts, DefaultAccount: defaultAccount, }) } return &domain.NotifiersResponse{ Notifiers: notifiers, }, nil } // storeNotification stores a clone of the notification in memory. Storing a // clone (rather than the caller's pointer) ensures the map never aliases a // notification that the caller, the queue, or a worker may still mutate. func (s *NotificationService) storeNotification(notification *domain.Notification) { s.mu.Lock() defer s.mu.Unlock() s.notifications[notification.ID] = notification.Clone() } // updateNotification updates a notification in memory with a clone of the // given notification, for the same reason as storeNotification. func (s *NotificationService) updateNotification(notification *domain.Notification) { s.mu.Lock() defer s.mu.Unlock() s.notifications[notification.ID] = notification.Clone() } // tenantCanAccess reports whether the caller identified by ctx is allowed to // see the given notification. Behavior: // - No auth context present (auth disabled): always allowed, preserving // pre-multi-tenant behavior. // - Auth context present with the admin role: always allowed. // - Auth context present without the admin role: allowed only if the // notification's ClientID matches the caller's. func (s *NotificationService) tenantCanAccess(ctx context.Context, notification *domain.Notification) bool { if notification == nil { return false } authCtx, ok := auth.GetAuthContext(ctx) if !ok || authCtx == nil { return true } if hasRole(authCtx.Roles, adminRole) { return true } return notification.ClientID == authCtx.ClientID } // hasRole reports whether role is present in roles. func hasRole(roles []string, role string) bool { for _, r := range roles { if r == role { return true } } return false } // checkAuthorization verifies that the caller is authorized to send to the given notifier/account. // Returns nil if authorized or if RBAC is not configured. func (s *NotificationService) checkAuthorization(ctx context.Context, notification *domain.Notification) error { if s.authz == nil || !s.authz.HasRules() { return nil // RBAC not configured } authCtx, ok := auth.GetAuthContext(ctx) if !ok { return nil // No auth context (auth may be disabled) } account := notification.Account if account == "" && s.accountResolver != nil { account = s.accountResolver.GetDefaultAccount(notification.Type) } if !s.authz.IsAuthorized(authCtx, notification.Type, account) { return fmt.Errorf("not authorized to send %s notifications to account %s", notification.Type, account) } return nil } // matchesFilter checks if a notification matches the filter func (s *NotificationService) matchesFilter(notification *domain.Notification, filter *domain.NotificationFilter) bool { if filter == nil { return true } // Check IDs if len(filter.IDs) > 0 { found := false for _, id := range filter.IDs { if notification.ID == id { found = true break } } if !found { return false } } // Check types if len(filter.Types) > 0 { found := false for _, t := range filter.Types { if notification.Type == t { found = true break } } if !found { return false } } // Check statuses if len(filter.Statuses) > 0 { found := false for _, s := range filter.Statuses { if notification.Status == s { found = true break } } if !found { return false } } // Check recipients if len(filter.Recipients) > 0 { found := false for _, fr := range filter.Recipients { for _, nr := range notification.Recipients { if fr == nr { found = true break } } if found { break } } if !found { return false } } // Check time ranges if filter.CreatedAfter != nil && notification.CreatedAt.Before(*filter.CreatedAfter) { return false } if filter.CreatedBefore != nil && notification.CreatedAt.After(*filter.CreatedBefore) { return false } return true }