notifier/internal/service/service.go

package service

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/igodwin/notifier/internal/domain"
)

// AccountResolver is an interface for resolving default accounts
type AccountResolver interface {
	GetDefaultAccount(notifierType domain.NotificationType) string
}

// NotificationService implements the domain.NotificationService interface
type NotificationService struct {
	factory         domain.NotifierFactory
	queue           domain.Queue
	accountResolver AccountResolver
	notifications   map[string]*domain.Notification
	mu              sync.RWMutex
	workerCount     int
	stopChan        chan struct{}
	wg              sync.WaitGroup
}

// NewNotificationService creates a new notification service
func NewNotificationService(factory domain.NotifierFactory, queue domain.Queue, workerCount int, accountResolver AccountResolver) *NotificationService {
	if workerCount <= 0 {
		workerCount = 10
	}

	return &NotificationService{
		factory:         factory,
		queue:           queue,
		accountResolver: accountResolver,
		notifications:   make(map[string]*domain.Notification),
		workerCount:     workerCount,
		stopChan:        make(chan struct{}),
	}
}

// Start starts the worker pool
func (s *NotificationService) Start(ctx context.Context) error {
	for i := 0; i < s.workerCount; i++ {
		s.wg.Add(1)
		go s.worker(ctx, i)
	}
	return nil
}

// Stop stops the service gracefully
func (s *NotificationService) Stop() error {
	close(s.stopChan)
	s.wg.Wait()
	return s.queue.Close()
}

// worker processes notifications from the queue
func (s *NotificationService) worker(ctx context.Context, id 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) {
	notification := msg.Notification

	// 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 {
		notification.Status = domain.StatusFailed
		notification.LastError = fmt.Sprintf("failed to create notifier: %v", err)
		s.queue.Nack(ctx, msg.ID, false)
		s.updateNotification(notification)
		return
	}

	// Send the notification
	result, err := notifier.Send(ctx, notification)
	if err != nil || !result.Success {
		notification.RetryCount++
		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.queue.Nack(ctx, msg.ID, true) // Requeue
		} else {
			notification.Status = domain.StatusFailed
			s.queue.Nack(ctx, msg.ID, false) // Don't requeue
		}
	} else {
		notification.Status = domain.StatusSent
		now := time.Now()
		notification.SentAt = &now
		s.queue.Ack(ctx, msg.ID)
	}

	s.updateNotification(notification)
}

// Send queues a notification for delivery
func (s *NotificationService) Send(ctx context.Context, notification *domain.Notification) (*domain.NotificationResult, error) {
	// Store the notification
	s.storeNotification(notification)

	// Enqueue for processing
	if err := s.queue.Enqueue(ctx, notification); err != nil {
		return &domain.NotificationResult{
			NotificationID: notification.ID,
			Success:        false,
			Error:          fmt.Sprintf("failed to enqueue: %v", err),
			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))

	// Store all notifications
	for _, notification := range notifications {
		s.storeNotification(notification)
	}

	// Enqueue batch
	if err := s.queue.EnqueueBatch(ctx, notifications); 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
func (s *NotificationService) GetNotification(ctx context.Context, id string) (*domain.Notification, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	notification, exists := s.notifications[id]
	if !exists {
		return nil, fmt.Errorf("notification not found: %s", id)
	}

	return notification, nil
}

// ListNotifications retrieves notifications matching the filter
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.matchesFilter(notification, filter) {
			results = append(results, notification)
		}
	}

	// 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
func (s *NotificationService) CancelNotification(ctx context.Context, id string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	notification, exists := s.notifications[id]
	if !exists {
		return fmt.Errorf("notification not found: %s", id)
	}

	if notification.Status == domain.StatusSent {
		return fmt.Errorf("notification already sent")
	}

	notification.Status = domain.StatusFailed
	notification.LastError = "cancelled by user"

	return nil
}

// RetryNotification retries a failed notification
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(),
		}, fmt.Errorf("notification already sent")
	}

	// Reset retry count and status
	notification.RetryCount = 0
	notification.Status = domain.StatusPending

	// Re-enqueue
	return s.Send(ctx, notification)
}

// GetStats returns notification statistics
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 {
		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
}

// storeNotification stores a notification in memory
func (s *NotificationService) storeNotification(notification *domain.Notification) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.notifications[notification.ID] = notification
}

// updateNotification updates a notification in memory
func (s *NotificationService) updateNotification(notification *domain.Notification) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.notifications[notification.ID] = notification
}

// 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
}