From d486a5e44ffb350f4192303d931ca6842698109a Mon Sep 17 00:00:00 2001 From: Mazyar Date: Mon, 22 Jun 2026 12:53:22 +0330 Subject: [PATCH] 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. --- cmd/worker/bootstrap/bootstrap.go | 23 +++++++++++++++++ cmd/worker/bootstrap/config.go | 2 ++ cmd/worker/workers/notification.go | 40 +++++++++++++++++++++++++++++ internal/notification/repository.go | 36 ++++++++++++++++++++++++++ internal/notification/service.go | 21 +++++++++++++++ 5 files changed, 122 insertions(+) create mode 100644 cmd/worker/workers/notification.go diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index 43156e1..f8c317b 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -6,6 +6,7 @@ import ( "time" "tm/cmd/worker/workers" "tm/internal/notice" + notificationDomain "tm/internal/notification" "tm/internal/tender" ai_summarizer "tm/pkg/ai_summarizer" "tm/pkg/config" @@ -87,10 +88,12 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(), "translation_enabled": config.Worker.TranslationEnabled, "translation_interval": config.Worker.TranslationInterval, + "notification_interval": config.Worker.NotificationInterval, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) tenderRepo := tender.NewRepository(mongoManager, appLogger) + notificationRepo := notificationDomain.NewRepository(mongoManager, appLogger) // Create a single shared cron scheduler for all recurring jobs scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) @@ -159,6 +162,26 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{}) } + notificationInterval := config.Worker.NotificationInterval + if notificationInterval == "" { + notificationInterval = "0 * * * * *" + appLogger.Warn("WORKER_NOTIFICATION_INTERVAL not set, using default schedule", map[string]interface{}{ + "interval": notificationInterval, + }) + } + + scheduler.AddJob(schedule.Job{ + Name: "Scheduled Notification Worker Job", + Func: func() { + worker := workers.NewNotificationWorker(notificationRepo, appLogger) + worker.Run() + }, + Expr: notificationInterval, + }) + appLogger.Info("Scheduled notification delivery worker", map[string]interface{}{ + "interval": notificationInterval, + }) + // Kick off one notice-processing pass without blocking startup (cron continues on schedule) go func() { w := workers.NewNoticeWorker( diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 35c3a24..4a7bb6f 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -33,6 +33,8 @@ type WorkerConfig struct { // TranslationEnabled schedules the automatic batch translation cron job on the worker. // On-demand translation (admin/public tender endpoints and AI pipeline routes on web) is unaffected. TranslationEnabled bool `env:"WORKER_TRANSLATION_ENABLED" envDefault:"true"` + // NotificationInterval schedules promotion of due scheduled notifications from pending to sent. + NotificationInterval string `env:"WORKER_NOTIFICATION_INTERVAL" envDefault:"0 * * * * *"` } // AISummarizerConfig holds configuration for the external AI summarizer service. diff --git a/cmd/worker/workers/notification.go b/cmd/worker/workers/notification.go new file mode 100644 index 0000000..9188a87 --- /dev/null +++ b/cmd/worker/workers/notification.go @@ -0,0 +1,40 @@ +package workers + +import ( + "context" + "time" + + "tm/internal/notification" + "tm/pkg/logger" +) + +// NotificationWorker promotes due scheduled notifications from pending to sent. +type NotificationWorker struct { + Repository notification.Repository + Logger logger.Logger +} + +// NewNotificationWorker creates a notification delivery worker. +func NewNotificationWorker(repository notification.Repository, logger logger.Logger) *NotificationWorker { + return &NotificationWorker{ + Repository: repository, + Logger: logger, + } +} + +// Run marks pending scheduled notifications as sent once their delivery time has passed. +func (w *NotificationWorker) Run() { + count, err := w.Repository.MarkDueScheduledAsSent(context.Background(), time.Now().Unix()) + if err != nil { + w.Logger.Error("Scheduled notification worker failed", map[string]interface{}{ + "error": err.Error(), + }) + return + } + + if count > 0 { + w.Logger.Info("Scheduled notification worker completed", map[string]interface{}{ + "count": count, + }) + } +} diff --git a/internal/notification/repository.go b/internal/notification/repository.go index a1fa23a..4095935 100644 --- a/internal/notification/repository.go +++ b/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 +} diff --git a/internal/notification/service.go b/internal/notification/service.go index f98fff4..7f89a11 100644 --- a/internal/notification/service.go +++ b/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, + }) + } +}