package main import ( "context" "fmt" "os" "os/signal" "syscall" "github.com/hibiken/asynq" "github.com/rs/zerolog/log" "github.com/treytartt/casera-api/internal/config" "github.com/treytartt/casera-api/internal/database" "github.com/treytartt/casera-api/internal/push" "github.com/treytartt/casera-api/internal/repositories" "github.com/treytartt/casera-api/internal/services" "github.com/treytartt/casera-api/internal/worker/jobs" "github.com/treytartt/casera-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") } // 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) 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) 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") } // 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.TypeTaskReminder, jobHandler.HandleTaskReminder) mux.HandleFunc(jobs.TypeOverdueReminder, jobHandler.HandleOverdueReminder) mux.HandleFunc(jobs.TypeDailyDigest, jobHandler.HandleDailyDigest) mux.HandleFunc(jobs.TypeSendEmail, jobHandler.HandleSendEmail) mux.HandleFunc(jobs.TypeSendPush, jobHandler.HandleSendPush) // Start scheduler for periodic tasks scheduler := asynq.NewScheduler(redisOpt, nil) // Schedule task reminder notifications (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.TypeTaskReminder, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register task reminder job") } log.Info().Str("cron", "0 * * * *").Int("default_hour", cfg.Worker.TaskReminderHour).Msg("Registered task reminder job (runs hourly for per-user times)") // Schedule overdue reminder (runs every hour to support per-user custom times) if _, err := scheduler.Register("0 * * * *", asynq.NewTask(jobs.TypeOverdueReminder, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register overdue reminder job") } log.Info().Str("cron", "0 * * * *").Int("default_hour", cfg.Worker.OverdueReminderHour).Msg("Registered overdue reminder job (runs hourly for per-user times)") // Schedule daily digest (runs at configured hour - no per-user customization yet) dailyCron := formatCron(cfg.Worker.DailyNotifHour) if _, err := scheduler.Register(dailyCron, asynq.NewTask(jobs.TypeDailyDigest, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register daily digest job") } log.Info().Str("cron", dailyCron).Msg("Registered daily digest job") // 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") } // formatCron creates a cron expression for a specific hour (runs at minute 0) func formatCron(hour int) string { return fmt.Sprintf("0 %02d * * *", hour) }