diff --git a/apps/api/.env.dev.example b/apps/api/.env.dev.example index 036a4721..66b5994b 100644 --- a/apps/api/.env.dev.example +++ b/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= diff --git a/apps/api/cmd/BAT_worker/main.go b/apps/api/cmd/BAT_worker/main.go index 018123e6..6bdd404d 100644 --- a/apps/api/cmd/BAT_worker/main.go +++ b/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") diff --git a/apps/api/cmd/api/main.go b/apps/api/cmd/api/main.go index c8a7d88e..fd15a735 100644 --- a/apps/api/cmd/api/main.go +++ b/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) diff --git a/apps/api/internal/api/api.go b/apps/api/internal/api/api.go index 08e97cca..9001c779 100644 --- a/apps/api/internal/api/api.go +++ b/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) diff --git a/apps/api/internal/api/handlers/bat.go b/apps/api/internal/api/handlers/bat.go index 1c8a5377..dd796e8a 100644 --- a/apps/api/internal/api/handlers/bat.go +++ b/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) +} diff --git a/apps/api/internal/config/config.go b/apps/api/internal/config/config.go index e80d604f..fbb74fde 100644 --- a/apps/api/internal/config/config.go +++ b/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_"` diff --git a/apps/api/internal/db/queries/applications.sql b/apps/api/internal/db/queries/applications.sql index 6cab6457..4e70e8c5 100644 --- a/apps/api/internal/db/queries/applications.sql +++ b/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'; diff --git a/apps/api/internal/db/queries/event_roles.sql b/apps/api/internal/db/queries/event_roles.sql index 82e76ad2..91b715ca 100644 --- a/apps/api/internal/db/queries/event_roles.sql +++ b/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; \ No newline at end of file +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'; diff --git a/apps/api/internal/db/repository/application.go b/apps/api/internal/db/repository/application.go index cf1dd890..947d266c 100644 --- a/apps/api/internal/db/repository/application.go +++ b/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 } diff --git a/apps/api/internal/db/sqlc/applications.sql.go b/apps/api/internal/db/sqlc/applications.sql.go index 9b7bc37a..d541ba04 100644 --- a/apps/api/internal/db/sqlc/applications.sql.go +++ b/apps/api/internal/db/sqlc/applications.sql.go @@ -110,19 +110,6 @@ func (q *Queries) GetApplicationByUserAndEventID(ctx context.Context, arg GetApp return i, err } -const getTotalAcceptedApplicationsByEventId = `-- name: GetTotalAcceptedApplicationsByEventId :one -SELECT COUNT(*) FROM applications -WHERE event_id = $1::uuid - AND status = 'accepted' -` - -func (q *Queries) GetTotalAcceptedApplicationsByEventId(ctx context.Context, eventID uuid.UUID) (int64, error) { - row := q.db.QueryRow(ctx, getTotalAcceptedApplicationsByEventId, eventID) - var count int64 - err := row.Scan(&count) - return count, err -} - const joinWaitlist = `-- name: JoinWaitlist :exec UPDATE applications SET waitlist_join_time = COALESCE(waitlist_join_time, NOW()), @@ -313,6 +300,10 @@ SET waitlist_join_time = COALESCE(waitlist_join_time, NOW()), status = 'waitlisted' WHERE event_id = $1::uuid AND status = 'accepted' + AND user_id IN ( + SELECT user_id from event_roles AS er + WHERE er.role = 'applicant' +) ` func (q *Queries) TransitionAcceptedApplicationsToWaitlistByEventID(ctx context.Context, eventID uuid.UUID) error { diff --git a/apps/api/internal/db/sqlc/event_roles.sql.go b/apps/api/internal/db/sqlc/event_roles.sql.go index d7693126..0073dbe2 100644 --- a/apps/api/internal/db/sqlc/event_roles.sql.go +++ b/apps/api/internal/db/sqlc/event_roles.sql.go @@ -29,6 +29,19 @@ func (q *Queries) AssignRole(ctx context.Context, arg AssignRoleParams) error { return err } +const getAttendeeCountByEventId = `-- name: GetAttendeeCountByEventId :one +SELECT COUNT(*) FROM event_roles AS er +WHERE er.event_id = $1::uuid + AND er.role = 'attendee' +` + +func (q *Queries) GetAttendeeCountByEventId(ctx context.Context, eventID uuid.UUID) (int64, error) { + row := q.db.QueryRow(ctx, getAttendeeCountByEventId, eventID) + var count int64 + err := row.Scan(&count) + return count, err +} + const getEventStaff = `-- name: GetEventStaff :many SELECT u.id, u.name, u.email, u.email_verified, u.onboarded, u.image, u.created_at, u.updated_at, u.role, u.preferred_email, u.email_consent, er.role AS event_role FROM auth.users u diff --git a/apps/api/internal/services/application.go b/apps/api/internal/services/application.go index ba3b4785..6ceba220 100644 --- a/apps/api/internal/services/application.go +++ b/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()) diff --git a/apps/api/internal/services/bat.go b/apps/api/internal/services/bat.go index 885dcc63..748dc6db 100644 --- a/apps/api/internal/services/bat.go +++ b/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 +} diff --git a/apps/api/internal/services/email.go b/apps/api/internal/services/email.go index afe37f83..23196980 100644 --- a/apps/api/internal/services/email.go +++ b/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, diff --git a/apps/api/internal/tasks/bat.go b/apps/api/internal/tasks/bat.go index c54b6fca..28164ea4 100644 --- a/apps/api/internal/tasks/bat.go +++ b/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 +} diff --git a/apps/api/internal/workers/bat.go b/apps/api/internal/workers/bat.go index dc9c289a..4b954738 100644 --- a/apps/api/internal/workers/bat.go +++ b/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,7 +98,7 @@ 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 @@ -100,3 +106,10 @@ func (w *BATWorker) HandleTransitionWaitlistTask(ctx context.Context, t *asynq.T return nil } + +func (w *BATWorker) HandleShutdownScheduler(ctx context.Context, t *asynq.Task) error { + w.scheduler.Shutdown() + // Error returned by logging. + + return nil +}