Major waitlist transition fixes

apps/api/.env.dev.example

@@ -3,6 +3,11 @@ DATABASE_URL_MIGRATION="postgres://postgres:postgres@localhost:5432/coredb"
 REDIS_URL="redis://redis:6379"
 ALLOWED_ORIGINS="" # URLs in comma seperated list
 
+# Application waitlist
+MAX_ACCEPTED_APPLICATIONS=500
+ACCEPT_FROM_WAITLIST_COUNT=50
+ACCEPT_FROM_WAITLIST_PERIOD="@every 72h"
+
 # For OAuth
 AUTH_DISCORD_CLIENT_ID=
 AUTH_DISCORD_CLIENT_SECRET=

apps/api/cmd/BAT_worker/main.go

@@ -82,15 +82,16 @@ func main() {
 
 	sesClient := email.NewSESClient(cfg.AWS.AccessKey, cfg.AWS.AccessKeySecret, cfg.AWS.Region, logger)
 	emailService := services.NewEmailService(taskQueueClient, sesClient, logger)
-	batService := services.NewBatService(applicationRepo, eventRepo, userRepo, batRunsRepo, emailService, txm, nil, logger)
+	batService := services.NewBatService(applicationRepo, eventRepo, userRepo, batRunsRepo, emailService, txm, nil, scheduler, logger)
 	applicationService := services.NewApplicationService(applicationRepo, userRepo, eventService, emailService, txm, nil, nil, scheduler, logger)
 
-	BATWorker := workers.NewBATWorker(batService, applicationService, scheduler, logger)
+	BATWorker := workers.NewBATWorker(batService, applicationService, scheduler, taskQueueClient, logger)
 
 	mux := asynq.NewServeMux()
 	mux.HandleFunc(tasks.TypeCalculateAdmissions, BATWorker.HandleCalculateAdmissionsTask)
 	mux.HandleFunc(tasks.TypeTransitionWaitlist, BATWorker.HandleTransitionWaitlistTask)
 	mux.HandleFunc(tasks.TypeScheduleTransitionWaitlist, BATWorker.HandleScheduleTransitionWaitlistTask)
+	mux.HandleFunc(tasks.TypeShutdownScheduler, BATWorker.HandleShutdownScheduler)
 
 	if err := srv.Run(mux); err != nil {
 		logger.Fatal().Msg("Failed to run BAT worker")

apps/api/cmd/api/main.go

@@ -91,7 +91,7 @@ func main() {
 	emailService := services.NewEmailService(taskQueueClient, sesClient, logger)
 	applicationService := services.NewApplicationService(applicationRepo, userRepo, eventService, emailService, txm, r2Client, &cfg.CoreBuckets, nil, logger)
 	teamService := services.NewTeamService(teamRepo, teamMemberRepo, teamJoinRequestRepo, eventRepo, txm, logger)
-	batService := services.NewBatService(applicationRepo, eventRepo, userRepo, batRunsRepo, emailService, txm, taskQueueClient, logger)
+	batService := services.NewBatService(applicationRepo, eventRepo, userRepo, batRunsRepo, emailService, txm, taskQueueClient, nil, logger)
 
 	// Injections into handlers
 	apiHandlers := handlers.NewHandlers(authService, userService, eventInterestService, eventService, emailService, applicationService, teamService, batService, cfg, logger)

apps/api/internal/api/api.go

@@ -143,7 +143,8 @@ func (api *API) setupRoutes(mw *mw.Middleware) {
 			r.With(ensureEventAdmin).Post("/queue-confirmation-email", api.Handlers.Email.QueueConfirmationEmail)
 			r.With(ensureEventAdmin).Post("/calc-admissions", api.Handlers.Admission.HandleCalculateAdmissionsRequest)
 			r.With(ensureEventAdmin).Patch("/transition-waitlisted-applications", api.Handlers.Application.TransitionWaitlistedApplications)
-			r.With(ensureEventAdmin).Post("/queue-transition-waitlist-task", api.Handlers.Bat.QueueScheduleWaitlistTransitionTask)
+			r.With(ensureEventAdmin).Post("/begin-waitlist-transition", api.Handlers.Bat.QueueScheduleWaitlistTransitionTask)
+			r.With(ensureEventAdmin).Post("/shutdown-waitlist-scheduler", api.Handlers.Bat.QueueShutdownWaitlistSchedulerTask)
 			r.With(ensureEventAdmin).Post("/reviews/bat-runs/{runId}/release", api.Handlers.Admission.ReleaseDecisions)
 			r.With(ensureEventAdmin).Patch("/", api.Handlers.Event.UpdateEventById)
 			r.With(ensureEventAdmin).Post("/banner", api.Handlers.Event.UploadEventBanner)

apps/api/internal/api/handlers/bat.go

@@ -178,3 +178,21 @@ func (h *BatHandler) QueueScheduleWaitlistTransitionTask(w http.ResponseWriter,
 
 	res.Send(w, http.StatusCreated, nil)
 }
+
+//	 Queue Shutdown scheduler task
+//
+//		@Summary		Shutsdown an asynq scheduler
+//		@Description	Shutsdown the scheduler used for the waitlist transition task. Error returned through logs if a scheduler is not active.
+//		@Tags
+//
+//		@Success		200		"Scheduler shutdown successfully"
+//		@Failure		500		{object}	res.ErrorResponse	"Server error: failed to shutdown scheduler"
+//		@Router			/events/{eventId}/queue-transition-waitlist-task [post]
+func (h *BatHandler) QueueShutdownWaitlistSchedulerTask(w http.ResponseWriter, r *http.Request) {
+	err := h.BatService.QueueShutdownWaitlistScheduler()
+	if err != nil {
+		res.SendError(w, http.StatusInternalServerError, res.NewError("internal_err", "Failed to shutdown scheduler."))
+	}
+
+	res.Send(w, http.StatusOK, nil)
+}

apps/api/internal/config/config.go

@@ -55,14 +55,15 @@ type CoreBuckets struct {
 }
 
 type Config struct {
-	DatabaseURL                   string   `env:"DATABASE_URL"`
-	RedisURL                      string   `env:"REDIS_URL"`
-	Port                          string   `env:"PORT" envDefault:"8080"`
-	AllowedOriginsString          string   `env:"ALLOWED_ORIGINS"`
-	EmailTemplateDirectory        string   `env:"EMAIL_TEMPLATE_DIRECTORY"`
-	AllowedOrigins                []string ``
-	ApiOrigin                     string   `env:"API_ORIGIN"`
-	WaitlistWorkerSessionIdCookie string   `env:WAITLIST_WORKER_SESSION_ID_COOKIE`
+	DatabaseURL              string   `env:"DATABASE_URL"`
+	RedisURL                 string   `env:"REDIS_URL"`
+	Port                     string   `env:"PORT" envDefault:"8080"`
+	AllowedOriginsString     string   `env:"ALLOWED_ORIGINS"`
+	EmailTemplateDirectory   string   `env:"EMAIL_TEMPLATE_DIRECTORY"`
+	AllowedOrigins           []string ``
+	MaxAcceptedApplications  uint32   `env:"MAX_ACCEPTED_APPLICATIONS"`
+	AcceptFromWaitlistCount  uint32   `env:"ACCEPT_FROM_WAITLIST_COUNT"`
+	AcceptFromWaitlistPeriod string   `env:"ACCEPT_FROM_WAITLIST_PERIOD"`
 
 	Auth      AuthConfig   `envPrefix:"AUTH_"`
 	Cookie    CookieConfig `envPrefix:"COOKIE_"`

apps/api/internal/db/queries/applications.sql

@@ -104,7 +104,12 @@ UPDATE applications
 SET waitlist_join_time = COALESCE(waitlist_join_time, NOW()),
     status = 'waitlisted'
 WHERE event_id = @event_id::uuid
-  AND status = 'accepted';
+  AND status = 'accepted'
+  AND user_id IN (
+    SELECT user_id from event_roles AS er
+    WHERE er.role = 'applicant'
+)
+;
 
 -- name: TransitionWaitlistedApplicationsToAcceptedByEventID :many
 UPDATE applications
@@ -119,7 +124,3 @@ WHERE user_id IN (
 )
 RETURNING user_id;
 
--- name: GetTotalAcceptedApplicationsByEventId :one
-SELECT COUNT(*) FROM applications
-WHERE event_id = @event_id::uuid
-  AND status = 'accepted';

apps/api/internal/db/queries/event_roles.sql

@@ -24,4 +24,9 @@ WHERE er.event_id = $1;
 -- name: UpdateRole :exec
 UPDATE event_roles
 SET role = $3
-WHERE event_id = $1 AND user_id = $2;
+WHERE event_id = $1 AND user_id = $2;
+
+-- name: GetAttendeeCountByEventId :one
+SELECT COUNT(*) FROM event_roles AS er
+WHERE er.event_id = @event_id::uuid
+  AND er.role = 'attendee';

apps/api/internal/db/repository/application.go

@@ -211,7 +211,7 @@ func (r *ApplicationRepository) TransitionWaitlistedApplicationsToAcceptedByEven
 	})
 }
 
-func (r *ApplicationRepository) GetTotalAcceptedApplicationsByEventId(ctx context.Context, eventId uuid.UUID) (uint32, error) {
-	amount, err := r.db.Query.GetTotalAcceptedApplicationsByEventId(ctx, eventId)
+func (r *ApplicationRepository) GetAttendeeCountByEventId(ctx context.Context, eventId uuid.UUID) (uint32, error) {
+	amount, err := r.db.Query.GetAttendeeCountByEventId(ctx, eventId)
 	return uint32(amount), err
 }

apps/api/internal/services/application.go

@@ -577,20 +577,21 @@ func (s *ApplicationService) TransitionWaitlistedApplications(ctx context.Contex
 			return err
 		}
 
-		totalAccepted, err := s.appRepo.GetTotalAcceptedApplicationsByEventId(ctx, eventId)
+		attendeeCount, err := s.appRepo.GetAttendeeCountByEventId(ctx, eventId)
 		if err != nil {
 			s.logger.Err(err).Msg("Failed to get total accepted application amount.")
 		}
-		if (acceptanceQuota - totalAccepted) <= acceptanceCount {
-			s.logger.Info().Msgf("%v - %v <= %v", acceptanceQuota, totalAccepted, acceptanceCount)
+		if (acceptanceQuota - attendeeCount) <= acceptanceCount {
+			s.logger.Info().Msgf("Acceptance quota is close, shutting down waitlist transition scheduler. Remaining acceptances: %v - %v <= %v", acceptanceQuota, attendeeCount, acceptanceCount)
 			if s.scheduler != nil {
 				// The API also uses this file, and this function can be run from an endpoint so we have to check that the scheduler exists.
-				// Technically the task should be removed from the scheduler. However the scheduler is only running for this task.
+				// Technically the task should be removed from the scheduler via an scheduler ENTRY_ID. However the scheduler is only running for this task.
 				s.scheduler.Shutdown()
 			}
-			acceptanceCount = acceptanceQuota - totalAccepted
+			acceptanceCount = acceptanceQuota - attendeeCount
 		}
 
+		s.logger.Info().Msgf("Acceptance count: %v", acceptanceCount)
 		acceptedUserIds, err = txAppRepo.TransitionWaitlistedApplicationsToAcceptedByEventID(ctx, eventId, acceptanceCount)
 		if err != nil {
 			s.logger.Err(err).Msg(err.Error())

apps/api/internal/services/bat.go

@@ -47,12 +47,14 @@ type BatService struct {
 	emailService *EmailService
 	txm          *db.TransactionManager
 	taskQueue    *asynq.Client
+	scheduler    *asynq.Scheduler
 	logger       zerolog.Logger
 }
 
-func NewBatService(appRepo *repository.ApplicationRepository, eventRepo *repository.EventRepository, userRepo *repository.UserRepository, batRunsRepo *repository.BatRunsRepository, emailService *EmailService, txm *db.TransactionManager, taskQueue *asynq.Client, logger zerolog.Logger) *BatService {
+func NewBatService(appRepo *repository.ApplicationRepository, eventRepo *repository.EventRepository, userRepo *repository.UserRepository, batRunsRepo *repository.BatRunsRepository, emailService *EmailService, txm *db.TransactionManager, taskQueue *asynq.Client, scheduler *asynq.Scheduler, logger zerolog.Logger) *BatService {
 	return &BatService{
 		taskQueue:    taskQueue,
+		scheduler:    scheduler,
 		appRepo:      appRepo,
 		eventRepo:    eventRepo,
 		userRepo:     userRepo,
@@ -362,10 +364,10 @@ func (s *BatService) mapToCandidates(engine *bat.BatEngine, applications []sqlc.
 
 func (s *BatService) QueueScheduleWaitlistTransitionTask(ctx context.Context, eventId uuid.UUID) error {
 
-	// TODO: add period to config?
+	cfg := config.Load()
 	task, err := tasks.NewTaskScheduleTransitionWaitlist(tasks.ScheduleTransitionWaitlistPayload{
 		EventID: eventId,
-		Period:  "@every 72h",
+		Period:  cfg.AcceptFromWaitlistPeriod,
 	})
 	if err != nil {
 		s.logger.Err(err).Msg("Failed to create ScheduleTransitionWaitlist task")
@@ -381,3 +383,20 @@ func (s *BatService) QueueScheduleWaitlistTransitionTask(ctx context.Context, ev
 
 	return nil
 }
+
+func (s *BatService) QueueShutdownWaitlistScheduler() error {
+	task, err := tasks.NewTaskShutdownScheduler()
+	if err != nil {
+		s.logger.Err(err).Msg("Failed to create ShutdownWaitlistScheduler task")
+		return err
+	}
+
+	_, err = s.taskQueue.Enqueue(task, asynq.Queue("bat"))
+	if err != nil {
+		s.logger.Err(err).Msg("Failed to queue ShutdownWaitlistScheduler task")
+		return err
+	}
+	s.logger.Info().Msg("Queued ShutdownWaitlistScheduler task")
+
+	return nil
+}

apps/api/internal/services/email.go

@@ -82,6 +82,13 @@ func (s *EmailService) SendHtmlEmail(recipient string, subject string, name stri
 
 // TODO: refactor other queue functions to use a similar naming scheme
 func (s *EmailService) QueueSendHtmlEmailTask(to string, subject string, name string, templateFilePath string) (*asynq.TaskInfo, error) {
+	if len(to) == 0 {
+		s.logger.Warn().Msgf("No recipient email found for email being sent from template '%s'", templateFilePath)
+	}
+	if len(name) == 0 {
+		s.logger.Warn().Msgf("No recipient name found for email being sent from template '%s'", templateFilePath)
+	}
+
 	task, err := tasks.NewTaskSendHtmlEmail(tasks.SendHtmlEmailPayload{
 		To:               to,
 		Subject:          subject,

apps/api/internal/tasks/bat.go

@@ -11,6 +11,7 @@ const (
 	TypeCalculateAdmissions        = "admissions:calculate"
 	TypeScheduleTransitionWaitlist = "waitlist:scheduletransition"
 	TypeTransitionWaitlist         = "waitlist:transition"
+	TypeShutdownScheduler          = "waitlist:shutdownscheduler"
 )
 
 type CalculateAdmissionsPayload struct {
@@ -24,9 +25,9 @@ type ScheduleTransitionWaitlistPayload struct {
 }
 
 type TransitionWaitlistPayload struct {
-	EventID         uuid.UUID
-	AcceptanceCount uint32
-	AcceptanceQuota uint32
+	EventID                 uuid.UUID
+	AcceptFromWaitlistCount uint32
+	MaxAcceptedApplications uint32
 }
 
 func NewTaskCalculateAdmissions(payload CalculateAdmissionsPayload) (*asynq.Task, error) {
@@ -55,3 +56,7 @@ func NewTaskTransitionWaitlist(payload TransitionWaitlistPayload) (*asynq.Task,
 
 	return asynq.NewTask(TypeTransitionWaitlist, data), nil
 }
+
+func NewTaskShutdownScheduler() (*asynq.Task, error) {
+	return asynq.NewTask(TypeShutdownScheduler, nil), nil
+}

apps/api/internal/workers/bat.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/hibiken/asynq"
 	"github.com/rs/zerolog"
+	"github.com/swamphacks/core/apps/api/internal/config"
 	"github.com/swamphacks/core/apps/api/internal/db/sqlc"
 	"github.com/swamphacks/core/apps/api/internal/services"
 	"github.com/swamphacks/core/apps/api/internal/tasks"
@@ -22,15 +23,17 @@ type BATWorker struct {
 	batService         *services.BatService
 	applicationService *services.ApplicationService
 	scheduler          *asynq.Scheduler
+	taskQueue          *asynq.Client
 	logger             zerolog.Logger
 }
 
-func NewBATWorker(batService *services.BatService, applicationService *services.ApplicationService, scheduler *asynq.Scheduler, logger zerolog.Logger) *BATWorker {
+func NewBATWorker(batService *services.BatService, applicationService *services.ApplicationService, scheduler *asynq.Scheduler, taskQueue *asynq.Client, logger zerolog.Logger) *BATWorker {
 	return &BATWorker{
 		batService:         batService,
 		applicationService: applicationService,
 		logger:             logger.With().Str("worker", "BATWorker").Str("component", "BAT").Logger(),
 		scheduler:          scheduler,
+		taskQueue:          taskQueue,
 	}
 }
 
@@ -69,18 +72,21 @@ func (w *BATWorker) HandleScheduleTransitionWaitlistTask(ctx context.Context, t
 		return err
 	}
 
+	cfg := config.Load()
 	task, err := tasks.NewTaskTransitionWaitlist(tasks.TransitionWaitlistPayload{
-		EventID:         payload.EventID,
-		AcceptanceCount: 50,
-		AcceptanceQuota: 500,
+		EventID:                 payload.EventID,
+		AcceptFromWaitlistCount: cfg.AcceptFromWaitlistCount,
+		MaxAcceptedApplications: cfg.MaxAcceptedApplications,
 	})
 
 	w.scheduler.Start()
+	// The scheduler will make its first run after the period cycles once. So we queue our task immediately as well.
 	_, err = w.scheduler.Register(payload.Period, task, asynq.Queue("bat"))
 	if err != nil {
 		w.logger.Err(err)
 		return nil
 	}
+	_, err = w.taskQueue.Enqueue(task, asynq.Queue("bat"))
 
 	return nil
 }
@@ -92,11 +98,18 @@ func (w *BATWorker) HandleTransitionWaitlistTask(ctx context.Context, t *asynq.T
 		return err
 	}
 
-	err := w.applicationService.TransitionWaitlistedApplications(ctx, payload.EventID, payload.AcceptanceCount, payload.AcceptanceQuota)
+	err := w.applicationService.TransitionWaitlistedApplications(ctx, payload.EventID, payload.AcceptFromWaitlistCount, payload.MaxAcceptedApplications)
 	if err != nil {
 		w.logger.Err(err)
 		return nil
 	}
 
 	return nil
 }
+
+func (w *BATWorker) HandleShutdownScheduler(ctx context.Context, t *asynq.Task) error {
+	w.scheduler.Shutdown()
+	// Error returned by logging.
+
+	return nil
+}