package main import ( "context" "os" "os/signal" "syscall" "time" "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/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") } // 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, 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 reminderCron := formatCron(cfg.Worker.TaskReminderHour, cfg.Worker.TaskReminderMinute) if _, err := scheduler.Register(reminderCron, asynq.NewTask(jobs.TypeTaskReminder, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register task reminder job") } log.Info().Str("cron", reminderCron).Msg("Registered task reminder job") // Schedule overdue reminder at 9 AM UTC overdueCron := formatCron(cfg.Worker.OverdueReminderHour, 0) if _, err := scheduler.Register(overdueCron, asynq.NewTask(jobs.TypeOverdueReminder, nil)); err != nil { log.Fatal().Err(err).Msg("Failed to register overdue reminder job") } log.Info().Str("cron", overdueCron).Msg("Registered overdue reminder job") // Schedule daily digest at 11 AM UTC dailyCron := formatCron(cfg.Worker.DailyNotifHour, 0) 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 and minute (UTC) func formatCron(hour, minute int) string { return time.Date(0, 1, 1, hour, minute, 0, 0, time.UTC).Format("4 15 * * *") }