feat: add daily cron to apply scheduled default hours changes
Add ApplyScheduledDefaultHours which checks default_hours_scheduled_changes at midnight, bulk-updates working_hours with staged values, and inserts an admin_notification. Register as the 'apply-default-hours' job in the scheduler (daily at 00:05).
This commit is contained in:
@@ -2,10 +2,12 @@ package scheduling
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"crussell/clock"
|
||||
"crussell/db"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
@@ -253,3 +255,100 @@ func CleanupExpiredRefreshTokens(ctx context.Context) (int, error) {
|
||||
|
||||
return int(result.RowsAffected()), nil
|
||||
}
|
||||
|
||||
// ApplyScheduledDefaultHours applies any pending default hours changes that
|
||||
// have reached their effective_date. Runs daily at 00:05 to catch midnight
|
||||
// roll-overs even if the cron is slightly delayed.
|
||||
func ApplyScheduledDefaultHours(ctx context.Context) (int, error) {
|
||||
londonNow := clock.Now().In(clock.London)
|
||||
todayStr := londonNow.Format("2006-01-02")
|
||||
|
||||
tx, err := db.Conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to begin transaction: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := tx.Rollback(ctx); err != nil && !errors.Is(err, pgx.ErrTxClosed) {
|
||||
slog.Error("failed to rollback transaction", "err", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Find any changes where effective_date <= today (London) and not yet applied/cancelled
|
||||
var rowID int
|
||||
var hoursJSON string
|
||||
err = tx.QueryRow(ctx, `
|
||||
SELECT id, hours::text
|
||||
FROM default_hours_scheduled_changes
|
||||
WHERE effective_date <= $1::date
|
||||
AND applied_at IS NULL
|
||||
AND cancelled_at IS NULL
|
||||
LIMIT 1
|
||||
`, todayStr).Scan(&rowID, &hoursJSON)
|
||||
if err != nil {
|
||||
// No matching change — nothing to do
|
||||
return 0, tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// Parse the staged hours
|
||||
var hours []struct {
|
||||
Weekday int `json:"weekday"`
|
||||
StartTime string `json:"startTime"`
|
||||
EndTime string `json:"endTime"`
|
||||
IsOpen bool `json:"isOpen"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(hoursJSON), &hours); err != nil {
|
||||
return 0, fmt.Errorf("failed to parse hours JSON for change %d: %w", rowID, err)
|
||||
}
|
||||
|
||||
// Bulk-update working_hours with staged values
|
||||
weekdays := make([]int, len(hours))
|
||||
startTimes := make([]string, len(hours))
|
||||
endTimes := make([]string, len(hours))
|
||||
isOpenFlags := make([]bool, len(hours))
|
||||
for i, h := range hours {
|
||||
weekdays[i] = h.Weekday
|
||||
startTimes[i] = h.StartTime
|
||||
endTimes[i] = h.EndTime
|
||||
isOpenFlags[i] = h.IsOpen
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE working_hours AS wh
|
||||
SET start_time = v.start_time,
|
||||
end_time = v.end_time,
|
||||
is_open = v.is_open
|
||||
FROM (
|
||||
SELECT unnest($1::smallint[]) AS weekday,
|
||||
unnest($2::time[]) AS start_time,
|
||||
unnest($3::time[]) AS end_time,
|
||||
unnest($4::boolean[]) AS is_open
|
||||
) v
|
||||
WHERE wh.weekday = v.weekday
|
||||
`, weekdays, startTimes, endTimes, isOpenFlags); err != nil {
|
||||
return 0, fmt.Errorf("failed to update working_hours for change %d: %w", rowID, err)
|
||||
}
|
||||
|
||||
// Mark change as applied
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE default_hours_scheduled_changes
|
||||
SET applied_at = NOW()
|
||||
WHERE id = $1
|
||||
`, rowID); err != nil {
|
||||
return 0, fmt.Errorf("failed to mark change %d as applied: %w", rowID, err)
|
||||
}
|
||||
|
||||
// Insert admin notification
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO admin_notifications (reason, created_at)
|
||||
VALUES ('default_hours_changed', NOW())
|
||||
`); err != nil {
|
||||
return 0, fmt.Errorf("failed to insert notification for change %d: %w", rowID, err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return 0, fmt.Errorf("failed to commit: %w", err)
|
||||
}
|
||||
|
||||
slog.Info("Applied scheduled default hours change", "change_id", rowID, "effective_date", todayStr)
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
@@ -133,6 +133,16 @@ func RegisterAll(s *Scheduler) {
|
||||
Handler: scheduling.CleanupOldNameHistory,
|
||||
})
|
||||
|
||||
// === STAGED HOURS CHANGE ===
|
||||
|
||||
s.Register(Job{
|
||||
Name: "apply-default-hours",
|
||||
Schedule: "5 0 * * *", // Daily at 00:05 — after midnight to avoid race
|
||||
Timeout: 30 * time.Second,
|
||||
Concurrency: 1,
|
||||
Handler: scheduling.ApplyScheduledDefaultHours,
|
||||
})
|
||||
|
||||
// === BUSINESS LOGIC JOBS ===
|
||||
|
||||
s.Register(Job{
|
||||
|
||||
@@ -407,17 +407,14 @@ func TestColoredRows_Reset(t *testing.T) {
|
||||
// RegisterAll Tests
|
||||
// ============================================================
|
||||
|
||||
// TestRegisterAll_RegistersExpectedJobs verifies RegisterAll registers exactly 19 jobs
|
||||
// with all required fields populated (non-empty Name, non-empty Schedule, non-nil
|
||||
// Handler, positive Timeout).
|
||||
func TestRegisterAll_RegistersExpectedJobs(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := New()
|
||||
RegisterAll(s)
|
||||
|
||||
if got := len(s.registry); got != 19 {
|
||||
t.Fatalf("RegisterAll() registered %d jobs, want 19", got)
|
||||
if got := len(s.registry); got != 20 {
|
||||
t.Fatalf("RegisterAll() registered %d jobs, want 20", got)
|
||||
}
|
||||
|
||||
registered := make(map[string]Job, len(s.registry))
|
||||
@@ -490,6 +487,7 @@ func expectedJobNames() map[string]bool {
|
||||
"transition-discount-campaigns": true,
|
||||
"cleanup-verification-codes": true,
|
||||
"cleanup-refresh-tokens": true,
|
||||
"apply-default-hours": true,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user