Enhance notification delivery system with scheduled processing

- Introduced a new `NotificationWorker` to promote due scheduled notifications from pending to sent, improving notification management.
- Added `NotificationInterval` configuration to schedule the notification delivery worker, with a default value for flexibility.
- Implemented `MarkDueScheduledAsSent` method in the notification repository to update the status of notifications based on their delivery time.
- Updated the notification service to process due scheduled notifications during relevant operations, ensuring timely delivery.

This update enhances the notification system by automating the delivery of scheduled notifications, improving user engagement and operational efficiency.

internal/notification/repository.go

@@ -58,6 +58,7 @@ type Repository interface {
 	Update(ctx context.Context, notification *Notification) error
 	MarkAsSeen(ctx context.Context, notificationID, userID string) error
 	MarkAllAsSeen(ctx context.Context, userID string) error
+	MarkDueScheduledAsSent(ctx context.Context, dueBefore int64) (int64, error)
 }
 
 type notificationRepository struct {
@@ -74,6 +75,11 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Re
 		*orm.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}),
 		*orm.NewIndex("seen_idx", bson.D{{Key: "seen", Value: 1}}),
 		*orm.NewIndex("created_at_idx", bson.D{{Key: "created_at", Value: -1}}),
+		*orm.NewIndex("scheduled_delivery_idx", bson.D{
+			{Key: "status", Value: 1},
+			{Key: "is_scheduled", Value: 1},
+			{Key: "schedule_at", Value: 1},
+		}),
 	}
 
 	err := mongoManager.CreateIndexes("notifications", indexes)
@@ -279,3 +285,33 @@ func (r *notificationRepository) MarkAllAsSeen(ctx context.Context, userID strin
 
 	return nil
 }
+
+// MarkDueScheduledAsSent updates pending scheduled notifications whose delivery time has passed.
+func (r *notificationRepository) MarkDueScheduledAsSent(ctx context.Context, dueBefore int64) (int64, error) {
+	filter := bson.M{
+		"status":       string(DeliveryStatusPending),
+		"is_scheduled": true,
+		"schedule_at": bson.M{
+			"$gt":  0,
+			"$lte": dueBefore,
+		},
+	}
+
+	update := bson.M{
+		"$set": bson.M{
+			"status":     string(DeliveryStatusSent),
+			"updated_at": time.Now().Unix(),
+		},
+	}
+
+	result, err := r.collection.UpdateMany(ctx, filter, update)
+	if err != nil {
+		r.logger.Error("Failed to mark due scheduled notifications as sent", map[string]interface{}{
+			"error":      err.Error(),
+			"due_before": dueBefore,
+		})
+		return 0, err
+	}
+
+	return result.ModifiedCount, nil
+}

internal/notification/service.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"slices"
 	"strings"
+	"time"
 
 	"tm/internal/customer"
 	"tm/internal/user"
@@ -186,6 +187,8 @@ func (s *notificationService) persistNotification(ctx context.Context, recipient
 }
 
 func (s *notificationService) GetNotifications(ctx context.Context, req *SearchForm, pagination *response.Pagination) (*NotificationListResponse, error) {
+	s.processDueScheduled(ctx)
+
 	search := req.ResolvedSearch()
 
 	s.logger.Info("Getting notifications", map[string]interface{}{
@@ -264,6 +267,8 @@ func (s *notificationService) GetNotifications(ctx context.Context, req *SearchF
 }
 
 func (s *notificationService) GetNotification(ctx context.Context, notificationID string) (*NotificationResponse, error) {
+	s.processDueScheduled(ctx)
+
 	s.logger.Info("Getting notification", map[string]interface{}{
 		"notification_id": notificationID,
 	})
@@ -467,3 +472,19 @@ func (s *notificationService) getCustomers(ctx context.Context, values []string,
 
 	return recipients, nil
 }
+
+func (s *notificationService) processDueScheduled(ctx context.Context) {
+	count, err := s.repository.MarkDueScheduledAsSent(ctx, time.Now().Unix())
+	if err != nil {
+		s.logger.Error("Failed to process due scheduled notifications", map[string]interface{}{
+			"error": err.Error(),
+		})
+		return
+	}
+
+	if count > 0 {
+		s.logger.Info("Processed due scheduled notifications", map[string]interface{}{
+			"count": count,
+		})
+	}
+}