package main import ( "context" "os" "os/signal" "syscall" "github.com/hibiken/asynq" "github.com/redis/go-redis/v9" "github.com/rs/zerolog/log" "github.com/treytartt/honeydue-api/internal/config" "github.com/treytartt/honeydue-api/internal/database" "github.com/treytartt/honeydue-api/internal/monitoring" "github.com/treytartt/honeydue-api/internal/push" "github.com/treytartt/honeydue-api/internal/repositories" "github.com/treytartt/honeydue-api/internal/services" "github.com/treytartt/honeydue-api/internal/worker/jobs" "github.com/treytartt/honeydue-api/pkg/utils" ) func main() { // Initialize logger utils.InitLogger(true) // Load configuration cfg, err := config.Load() if err != nil { log.Fatal().Err(err).Msg("Failed to load configuration") } // Check worker kill switch if !cfg.Features.WorkerEnabled { log.Warn().Msg("Worker disabled by FEATURE_WORKER_ENABLED=false — exiting") os.Exit(0) } // Initialize database db, err := database.Connect(&cfg.Database, cfg.Server.Debug) if err != nil { log.Fatal().Err(err).Msg("Failed to connect to database") } log.Info().Msg("Connected to database") // Get underlying *sql.DB for cleanup sqlDB, _ := db.DB() defer sqlDB.Close() // Initialize push client (APNs + FCM) var pushClient *push.Client pushClient, err = push.NewClient(&cfg.Push, cfg.Features.PushEnabled) if err != nil { log.Warn().Err(err).Msg("Failed to initialize push client - push notifications disabled") } else { log.Info(). Bool("ios_enabled", pushClient.IsIOSEnabled()). Bool("android_enabled", pushClient.IsAndroidEnabled()). Msg("Push notification client initialized") } // Initialize email service (optional) var emailService *services.EmailService if cfg.Email.Host != "" { emailService = services.NewEmailService(&cfg.Email, cfg.Features.EmailEnabled) log.Info().Str("host", cfg.Email.Host).Msg("Email service initialized") } // Initialize notification service for actionable push notifications notificationRepo := repositories.NewNotificationRepository(db) notificationService := services.NewNotificationService(notificationRepo, pushClient) log.Info().Msg("Notification service initialized") // Parse Redis URL for Asynq redisOpt, err := asynq.ParseRedisURI(cfg.Redis.URL) if err != nil { log.Fatal().Err(err).Msg("Failed to parse Redis URL") } // Initialize monitoring service (if Redis is available) var monitoringService *monitoring.Service redisClientOpt, ok := redisOpt.(asynq.RedisClientOpt) if ok { redisClient := redis.NewClient(&redis.Options{ Addr: redisClientOpt.Addr, Password: redisClientOpt.Password, DB: redisClientOpt.DB, }) // Verify Redis connection if err := redisClient.Ping(context.Background()).Err(); err != nil { log.Warn().Err(err).Msg("Failed to connect to Redis for monitoring - monitoring disabled") } else { monitoringService = monitoring.NewService(monitoring.Config{ Process: "worker", RedisClient: redisClient, DB: db, // Pass database for enable_monitoring setting sync }) // Reinitialize logger with monitoring writer utils.InitLoggerWithWriter(cfg.Server.Debug, monitoringService.LogWriter()) // Create Asynq inspector for queue statistics inspector := asynq.NewInspector(redisOpt) monitoringService.SetAsynqInspector(inspector) // Start stats collection monitoringService.Start() defer monitoringService.Stop() log.Info(). Bool("log_capture_enabled", monitoringService.IsEnabled()). Msg("Monitoring service initialized") } } // Create Asynq server srv := asynq.NewServer( redisOpt, asynq.Config{ Concurrency: 10, Queues: map[string]int{ "critical": 6, "default": 3, "low": 1, }, ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, task *asynq.Task, err error) { log.Error(). Err(err). Str("type", task.Type()). Bytes("payload", task.Payload()). Msg("Task processing failed") }), }, ) // Create job handler jobHandler := jobs.NewHandler(db, pushClient, emailService, notificationService, cfg) // Create Asynq mux and register handlers mux := asynq.NewServeMux() mux.HandleFunc(jobs.TypeSmartReminder, jobHandler.HandleSmartReminder) mux.HandleFunc(jobs.TypeDailyDigest, jobHandler.HandleDailyDigest) mux.HandleFunc(jobs.TypeSendEmail, jobHandler.HandleSendEmail) mux.HandleFunc(jobs.TypeSendPush, jobHandler.HandleSendPush) mux.HandleFunc(jobs.TypeOnboardingEmails, jobHandler.HandleOnboardingEmails) mux.HandleFunc(jobs.TypeReminderLogCleanup, jobHandler.HandleReminderLogCleanup) // Register email job handlers (welcome, verification, password reset, password changed) if emailService != nil { emailJobHandler := jobs.NewEmailJobHandler(emailService) emailJobHandler.RegisterHandlers(mux) } // Start scheduler for periodic tasks scheduler := asynq.NewScheduler(redisOpt, nil) // Schedule smart reminder notifications (runs every hour to support per-user custom times) // Replaces old task reminder and overdue reminder with frequency-aware system // Uses TaskReminderLog to prevent duplicate notifications if _, err := scheduler.Register("0 * * * *", asynq.NewTask(jobs.TypeSmartReminder, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register smart reminder job") } log.Info().Str("cron", "0 * * * *").Int("default_hour", cfg.Worker.TaskReminderHour).Msg("Registered smart reminder job (runs hourly for per-user times)") // Schedule daily digest (runs every hour to support per-user custom times) // The job handler filters users based on their preferred notification hour if _, err := scheduler.Register("0 * * * *", asynq.NewTask(jobs.TypeDailyDigest, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register daily digest job") } log.Info().Str("cron", "0 * * * *").Int("default_hour", cfg.Worker.DailyNotifHour).Msg("Registered daily digest job (runs hourly for per-user times)") // Schedule onboarding emails (runs daily at 10:00 AM UTC) // Sends emails to users who haven't created residences or tasks after registration if _, err := scheduler.Register("0 10 * * *", asynq.NewTask(jobs.TypeOnboardingEmails, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register onboarding emails job") } log.Info().Str("cron", "0 10 * * *").Msg("Registered onboarding emails job (runs daily at 10:00 AM UTC)") // Schedule reminder log cleanup (runs daily at 3:00 AM UTC) // Removes reminder logs older than 90 days to prevent table bloat if _, err := scheduler.Register("0 3 * * *", asynq.NewTask(jobs.TypeReminderLogCleanup, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register reminder log cleanup job") } log.Info().Str("cron", "0 3 * * *").Msg("Registered reminder log cleanup job (runs daily at 3:00 AM UTC)") // Handle graceful shutdown quit := make(chan os.Signal, 1) signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) // Start scheduler in goroutine go func() { if err := scheduler.Run(); err != nil { log.Fatal().Err(err).Msg("Failed to start scheduler") } }() // Start worker server in goroutine go func() { log.Info().Msg("Starting worker server...") if err := srv.Run(mux); err != nil { log.Fatal().Err(err).Msg("Failed to start worker server") } }() <-quit log.Info().Msg("Shutting down worker...") // Graceful shutdown srv.Shutdown() scheduler.Shutdown() log.Info().Msg("Worker stopped") }