Skip to content

feat: add batch process for updating status of classes and enrolled_at per ticket 607 #1164

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
carddev81 wants to merge 1 commit into
base: main
Choose a base branch
from
+157 −9
Open

feat: add batch process for updating status of classes and enrolled_at per ticket 607 #1164

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 17 additions & 9 deletions backend/src/models/jobs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@ func (cj *CronJob) BeforeCreate(tx *gorm.DB) error {
cj.Category = OpenContentJob
} else if slices.Contains(AllDefaultProviderJobs, JobType(cj.Name)) {
cj.Category = ProviderPlatformJob
} else if slices.Contains(AllSystemJobs, JobType(cj.Name)) {
cj.Category = SystemJob
}
switch cj.Name {
case string(RetryVideoDownloadsJob):
Expand All @@ -41,6 +43,8 @@ func (cj *CronJob) BeforeCreate(tx *gorm.DB) error {
schedule = EveryDaytimeHour
}
cj.Schedule = schedule
case string(ActivateScheduledClassesJob):
cj.Schedule = EveryMorningAt5AM
default:
cj.Schedule = os.Getenv("MIDDLEWARE_CRON_SCHEDULE")
}
Expand Down Expand Up @@ -83,24 +87,28 @@ func (RunnableTask) TableName() string { return "runnable_tasks" }
const (
ProviderPlatformJob = 1
OpenContentJob = 2
SystemJob = 3

GetMilestonesJob JobType = "get_milestones"
GetCoursesJob JobType = "get_courses"
GetActivityJob JobType = "get_activity"

ScrapeKiwixJob JobType = "scrape_kiwix"
RetryVideoDownloadsJob JobType = "retry_video_downloads"
RetryManualDownloadJob JobType = "retry_manual_download"
SyncVideoMetadataJob JobType = "sync_video_metadata"
AddVideosJob JobType = "add_videos"
EveryDaytimeHour string = "0 6-20 * * *"
EverySundayAt8PM string = "0 20 * * 6"
StatusPending JobStatus = "pending"
StatusRunning JobStatus = "running"
ScrapeKiwixJob JobType = "scrape_kiwix"
RetryVideoDownloadsJob JobType = "retry_video_downloads"
RetryManualDownloadJob JobType = "retry_manual_download"
SyncVideoMetadataJob JobType = "sync_video_metadata"
AddVideosJob JobType = "add_videos"
ActivateScheduledClassesJob JobType = "activate_scheduled_classes"
EveryDaytimeHour string = "0 6-20 * * *"
EverySundayAt8PM string = "0 20 * * 6"
EveryMorningAt5AM string = "0 5 * * *"
StatusPending JobStatus = "pending"
StatusRunning JobStatus = "running"
)

var AllDefaultProviderJobs = []JobType{GetCoursesJob, GetMilestonesJob, GetActivityJob}
var AllContentProviderJobs = []JobType{ScrapeKiwixJob, RetryVideoDownloadsJob, SyncVideoMetadataJob}
var AllSystemJobs = []JobType{ActivateScheduledClassesJob}

func (jt JobType) IsVideoJob() bool {
switch jt {
Expand Down
45 changes: 45 additions & 0 deletions backend/src/tasks/scheduler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,54 @@ func (s *Scheduler) generateTasks() ([]models.RunnableTask, error) {
} else {
log.Println("Failed to generate provider tasks")
}
if systemTasks, err := s.generateSystemTasks(); err == nil {
allTasks = append(allTasks, systemTasks...)
} else {
log.Println("Failed to generate system tasks")
}
return allTasks, nil
}

// generateSystemTasks builds the platform-wide batch tasks that are not tied to
// any provider (e.g. activating classes whose scheduled start date has arrived).
func (s *Scheduler) generateSystemTasks() ([]models.RunnableTask, error) {
systemTasks := make([]models.RunnableTask, 0, len(models.AllSystemJobs))
for _, jobType := range models.AllSystemJobs {
created, err := s.createIfNotExists(jobType)
if err != nil {
log.Errorf("failed to create system job %s: %v", jobType, err)
return nil, err
}
task := models.RunnableTask{JobID: created.ID, Status: models.StatusPending}
if err := s.intoSystemTask(created, &task); err != nil {
log.Errorf("failed to create task for system job %s: %v", jobType, err)
return nil, err
}
systemTasks = append(systemTasks, task)
}
return systemTasks, nil
}

// intoSystemTask finds or creates the single provider-less RunnableTask for a
// system job and prepares its parameters for publishing.
func (s *Scheduler) intoSystemTask(cj *models.CronJob, task *models.RunnableTask) error {
if task.ID == 0 {
if err := s.db.Model(&models.RunnableTask{}).
Where("job_id = ? AND provider_platform_id IS NULL AND open_content_provider_id IS NULL", cj.ID).
FirstOrCreate(&task).Error; err != nil {
log.Errorf("failed to create task for job: %v. error: %v", cj.Name, err)
return err
}
if err := s.db.Model(&models.RunnableTask{}).Preload("Job").First(&task, task.ID).Error; err != nil {
log.Errorf("failed to reload task for job: %v. error: %v", cj.Name, err)
return err
}
}
task.Job = cj
task.Prepare(nil)
return nil
}

func (s *Scheduler) generateOpenContentProviderTasks() ([]models.RunnableTask, error) {
otherTasks := make([]models.RunnableTask, 0, 5)
providers := make([]models.OpenContentProvider, 0, 2)
Expand Down
1 change: 1 addition & 0 deletions provider-middleware/handlers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ func (sh *ServiceHandler) initSubscription() error {
{models.RetryVideoDownloadsJob.PubName(), sh.handleRetryFailedVideos},
{models.RetryManualDownloadJob.PubName(), sh.handleManualRetryDownload},
{models.SyncVideoMetadataJob.PubName(), sh.handleSyncVideoMetadata},
{models.ActivateScheduledClassesJob.PubName(), sh.handleActivateScheduledClasses},
}
for _, sub := range subscriptions {
timeout := CANCEL_TIMEOUT
Expand Down
94 changes: 94 additions & 0 deletions provider-middleware/program_classes.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
package main

import (
"UnlockEdv2/src/models"
"context"
"encoding/json"
"fmt"
"time"

"github.com/nats-io/nats.go"
"gorm.io/gorm"
)

// handleActivateScheduledClasses is the entrypoint for the daily
// `tasks.activate_scheduled_classes` job. It flips any class whose scheduled
// start date has is the current day of the job run (in its facility's local timezone) from Scheduled to
// Active. The status update goes through the same map-based update the user interface uses
// so that ProgramClass.AfterUpdate fires and backfills enrolled_at on the
// class's enrollments.
func (sh *ServiceHandler) handleActivateScheduledClasses(ctx context.Context, msg *nats.Msg) {
var body map[string]any
if err := json.Unmarshal(msg.Data, &body); err != nil {
logger().Errorf("failed to unmarshal activate_scheduled_classes message: %v", err)
return
}
jobId, ok := body["job_id"].(string)
if !ok {
logger().Errorf("job_id not found in activate_scheduled_classes message: %v", body)
return
}
success := sh.activateScheduledClasses(ctx) == nil
sh.cleanupJob(ctx, nil, jobId, success)
}

func (sh *ServiceHandler) activateScheduledClasses(ctx context.Context) error {
batchUserID, err := sh.systemBatchUserID(ctx)
if err != nil { //batch id doesn't exist then fail
logger().Errorf("cannot activate scheduled classes: %v", err)
return err
}

var classIDs []int
if err := sh.db.WithContext(ctx).
Model(&models.ProgramClass{}).
Joins("JOIN facilities f ON f.id = program_classes.facility_id").
Where("program_classes.status = ?", models.Scheduled).
Where("program_classes.archived_at IS NULL").
Where("program_classes.start_dt <= (now() AT TIME ZONE f.timezone)::date").
Pluck("program_classes.id", &classIDs).Error; err != nil {
logger().Errorf("failed to query scheduled classes to activate: %v", err)
return err
}

if len(classIDs) == 0 {
logger().Infoln("no scheduled classes are due for activation")
return nil
}
logger().Infof("activating %d scheduled class(es): %v", len(classIDs), classIDs)

batchCtx := context.WithValue(ctx, models.UserIDKey, batchUserID)
enrolledAt := time.Now().UTC()
if err := sh.db.WithContext(batchCtx).Transaction(func(tx *gorm.DB) error {
if err := tx.
Model(&models.ProgramClass{}).
Where("id IN ?", classIDs).
Set("class_ids", classIDs).
Updates(map[string]any{"status": models.Active}).Error; err != nil {
return err
Comment on lines +62 to +68

@coderabbitai coderabbitai Bot Jun 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: Search for ProgramClass AfterUpdate hook that might backfill enrolled_at

rg -n -C5 'func.*ProgramClass.*AfterUpdate' --type go

Repository: UnlockedLabs/UnlockEdv2

Length of output: 860

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# 1) Find where Set("class_ids", ...) is used
rg -n --hidden --no-ignore -S 'Set\("class_ids"\s*,\s*' .

# 2) Locate provider-middleware/program_classes.go if present and show around the targeted lines
# (Prefer exact filename match; otherwise search for the transaction pattern)
if [ -f "provider-middleware/program_classes.go" ]; then
  echo "---- provider-middleware/program_classes.go ----"
  nl -ba provider-middleware/program_classes.go | sed -n '40,100p'
else
  echo "---- searching for provider-middleware/program_classes.go ----"
  fd -t f -e go --hidden --no-ignore 'program_classes.go' -a . | head -n 50 || true
  # fallback: search for the transaction snippet structure
  rg -n --hidden --no-ignore -S 'WithContext\(batchCtx\)\.Transaction' . || true
fi

# 3) Show full AfterUpdate hook body (limited range around discovered location)
HOOK_FILE="backend/src/models/program_classes.go"
if [ -f "$HOOK_FILE" ]; then
  echo "---- $HOOK_FILE: AfterUpdate ----"
  # print from ~60 lines before hook to ~120 lines after (cap output)
  nl -ba "$HOOK_FILE" | sed -n '50,170p'
fi

Repository: UnlockedLabs/UnlockEdv2

Length of output: 331

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Print with line numbers using `cat -n` (no `nl` available in sandbox)
print_range() {
  local file="$1"
  local start="$2"
  local end="$3"
  echo "---- $file:$start-$end ----"
  cat -n "$file" | sed -n "${start},${end}p"
}

# provider-middleware: inspect around the Set("class_ids", ...)
print_range "provider-middleware/program_classes.go" 40 110

# model hook: inspect full AfterUpdate body
print_range "backend/src/models/program_classes.go" 60 170

# database/enrollments: context for another Set("class_ids", ...)
if [ -f "backend/src/database/class_enrollments.go" ]; then
  print_range "backend/src/database/class_enrollments.go" 230 330
fi

Repository: UnlockedLabs/UnlockEdv2

Length of output: 10652

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Find DatabaseFields definition and any hooks/callbacks that might auto-set update_user_id/update_user_id behavior
rg -n --hidden --no-ignore -S 'type DatabaseFields|func \\(.*DatabaseFields\\).*Before|func \\(.*DatabaseFields\\).*After|UpdateUserID|update_user_id' backend/src/models

# Find any hooks specifically on ProgramClassEnrollment for BeforeUpdate/AfterUpdate
rg -n --hidden --no-ignore -S 'type ProgramClassEnrollment|func \\(e \\*ProgramClassEnrollment\\) (BeforeUpdate|AfterUpdate|BeforeSave|AfterSave|BeforeCreate|AfterCreate)' backend/src/models

Repository: UnlockedLabs/UnlockEdv2

Length of output: 1107

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Show DatabaseFields implementation around line ~1-120
echo "---- backend/src/models/users.go: DatabaseFields ----"
cat -n backend/src/models/users.go | sed -n '1,130p'

# Search for ProgramClassEnrollment hooks beyond the one we saw (BeforeCreate)
echo "---- ProgramClassEnrollment hooks ----"
rg -n --hidden --no-ignore -S 'func \\(e \\*ProgramClassEnrollment\\) (BeforeUpdate|AfterUpdate|BeforeSave|AfterSave|BeforeCreate|AfterCreate|BeforeUpdate)' backend/src/models/program_classes.go || true

# Show ProgramClassEnrollment hook body section we already partially saw (extend range)
echo "---- backend/src/models/program_classes.go: ProgramClassEnrollment hooks ----"
cat -n backend/src/models/program_classes.go | sed -n '120,220p'

Repository: UnlockedLabs/UnlockEdv2

Length of output: 9199

🏁 Script executed:

#!/bin/bash
set -euo pipefail
# Print continuation of ProgramClassEnrollment.BeforeUpdate
cat -n backend/src/models/program_classes.go | sed -n '160,320p'

Repository: UnlockedLabs/UnlockEdv2

Length of output: 7052

🏁 Script executed:

#!/bin/bash
set -euo pipefail
# Print continuation of ProgramClassEnrollment.BeforeUpdate
cat -n backend/src/models/program_classes.go | sed -n '160,320p'

Repository: UnlockedLabs/UnlockEdv2

Length of output: 7052

Clarify Set("class_ids", classIDs) purpose and avoid overlapping enrolled_at backfill

  • Set("class_ids", classIDs) is required: models.ProgramClass.AfterUpdate calls tx.Get("class_ids") and uses it to update ProgramClassEnrollment rows when status changes to Active.
  • The AfterUpdate hook already backfills ProgramClassEnrollment.enrolled_at for enrollment_status = Enrolled where enrolled_at IS NULL, so the subsequent explicit ProgramClassEnrollment update in provider-middleware/program_classes.go (lines 70–78) is redundant for enrolled_at and may also skip setting update_user_id due to the enrolled_at IS NULL predicate running after the hook.
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@provider-middleware/program_classes.go` around lines 62 - 68, The
Set("class_ids", classIDs) call must remain because
models.ProgramClass.AfterUpdate reads tx.Get("class_ids") to drive enrollment
updates; remove or change the redundant ProgramClassEnrollment update in
provider-middleware/program_classes.go (the explicit update that backfills
enrolled_at and update_user_id after the Transaction) so it does not duplicate
the AfterUpdate hook's backfill for enrolled_at, and instead rely on the hook to
set enrolled_at for enrollment_status = Enrolled where enrolled_at IS NULL; if
you still need to update update_user_id for those rows, modify the hook
(models.ProgramClass.AfterUpdate) to also set update_user_id when it backfills
enrolled_at so the single path performs both updates atomically.

}
return tx.
Model(&models.ProgramClassEnrollment{}).
Where("class_id IN ?", classIDs).
Where("enrollment_status = ?", models.Enrolled).
Where("enrolled_at IS NULL").
Updates(map[string]any{
"enrolled_at": enrolledAt,
"update_user_id": batchUserID,
}).Error
}); err != nil {
logger().Errorf("failed to activate scheduled classes %v: %v", classIDs, err)
return err
}
return nil
}

func (sh *ServiceHandler) systemBatchUserID(ctx context.Context) (uint, error) {
var user models.User
if err := sh.db.WithContext(ctx).
Where("username = ?", "system_batch").
First(&user).Error; err != nil {
return 0, fmt.Errorf("system_batch user not found: %w", err)
}
return user.ID, nil
}
Loading