refactor: move auth, GDPR, and booking cleanup to centralized scheduler
Convert CleanupRevokedJTIs to return (int, error) and remove StartJTICleanup goroutine. Add CleanupStaleLoginEntries and CleanupGDPRExportCache for centralized scheduler. Add clock.London timezone location. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
+16
-30
@@ -8,8 +8,9 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"crussell/db"
|
|
||||||
"crussell/clock"
|
"crussell/clock"
|
||||||
|
"crussell/db"
|
||||||
|
|
||||||
"github.com/jackc/pgx/v5"
|
"github.com/jackc/pgx/v5"
|
||||||
|
|
||||||
"github.com/go-chi/jwtauth/v5"
|
"github.com/go-chi/jwtauth/v5"
|
||||||
@@ -43,7 +44,7 @@ func RevokeJTI(ctx context.Context, jti string, expiresAt time.Time) {
|
|||||||
}
|
}
|
||||||
tx, err := db.Conn.Begin(ctx)
|
tx, err := db.Conn.Begin(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("WARN: Failed to begin transaction for JTI revocation: %v\n", err)
|
log.Printf("WARN: Failed to begin transaction for JTI revocation: %v", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
defer tx.Rollback(ctx)
|
defer tx.Rollback(ctx)
|
||||||
@@ -54,12 +55,12 @@ func RevokeJTI(ctx context.Context, jti string, expiresAt time.Time) {
|
|||||||
jti, expiresAt)
|
jti, expiresAt)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// Log but don't fail - this is best effort
|
// Log but don't fail - this is best effort
|
||||||
fmt.Printf("WARN: Failed to revoke JTI %s: %v\n", jti, err)
|
log.Printf("WARN: Failed to revoke JTI %s: %v", jti, err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := tx.Commit(ctx); err != nil {
|
if err := tx.Commit(ctx); err != nil {
|
||||||
fmt.Printf("WARN: Failed to commit transaction for JTI revocation: %v\n", err)
|
log.Printf("WARN: Failed to commit transaction for JTI revocation: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -81,46 +82,31 @@ func IsJTIRevoked(ctx context.Context, jti string) bool {
|
|||||||
return exists
|
return exists
|
||||||
}
|
}
|
||||||
|
|
||||||
// CleanupRevokedJTIs removes expired entries from PostgreSQL
|
// CleanupRevokedJTIs removes expired entries from PostgreSQL and returns the count of deleted rows.
|
||||||
func CleanupRevokedJTIs(ctx context.Context) {
|
func CleanupRevokedJTIs(ctx context.Context) (int, error) {
|
||||||
if db.Conn == nil {
|
if db.Conn == nil {
|
||||||
return
|
return 0, nil
|
||||||
}
|
}
|
||||||
tx, err := db.Conn.Begin(ctx)
|
tx, err := db.Conn.Begin(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("WARN: Failed to begin transaction for JTI cleanup: %v\n", err)
|
log.Printf("WARN: Failed to begin transaction for JTI cleanup: %v", err)
|
||||||
return
|
return 0, err
|
||||||
}
|
}
|
||||||
defer tx.Rollback(ctx)
|
defer tx.Rollback(ctx)
|
||||||
|
|
||||||
_, err = tx.Exec(ctx,
|
tag, err := tx.Exec(ctx,
|
||||||
`DELETE FROM revoked_jtis WHERE expires_at < NOW()`)
|
`DELETE FROM revoked_jtis WHERE expires_at < NOW()`)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("WARN: Failed to cleanup revoked JTIs: %v\n", err)
|
log.Printf("WARN: Failed to cleanup revoked JTIs: %v", err)
|
||||||
return
|
return 0, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := tx.Commit(ctx); err != nil {
|
if err := tx.Commit(ctx); err != nil {
|
||||||
fmt.Printf("WARN: Failed to commit transaction for JTI cleanup: %v\n", err)
|
log.Printf("WARN: Failed to commit transaction for JTI cleanup: %v", err)
|
||||||
|
return 0, err
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
// StartJTICleanup starts a background goroutine to periodically clean expired JTIs
|
return int(tag.RowsAffected()), nil
|
||||||
func StartJTICleanup() {
|
|
||||||
go func() {
|
|
||||||
ticker := time.NewTicker(30 * time.Minute)
|
|
||||||
defer ticker.Stop()
|
|
||||||
for range ticker.C {
|
|
||||||
func() {
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
log.Printf("Panic recovered in JWT cleanup ticker: %v", r)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
CleanupRevokedJTIs(context.Background())
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func InitJWT(secret string) {
|
func InitJWT(secret string) {
|
||||||
|
|||||||
@@ -1,3 +1,7 @@
|
|||||||
|
// Package clock provides the single source of truth for time across the
|
||||||
|
// application. All wall-clock operations use Now() for UTC consistency with
|
||||||
|
// the database. LondonLocation provides Europe/London for cron schedules and
|
||||||
|
// scheduling calculations that depend on local business-closing rules.
|
||||||
package clock
|
package clock
|
||||||
|
|
||||||
import "time"
|
import "time"
|
||||||
@@ -8,3 +12,13 @@ import "time"
|
|||||||
func Now() time.Time {
|
func Now() time.Time {
|
||||||
return time.Now().UTC()
|
return time.Now().UTC()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// London is the Europe/London timezone location, used for cron schedules and
|
||||||
|
// business-closing calculations.
|
||||||
|
var London = func() *time.Location {
|
||||||
|
loc, err := time.LoadLocation("Europe/London")
|
||||||
|
if err != nil {
|
||||||
|
panic("clock: failed to load Europe/London timezone: " + err.Error())
|
||||||
|
}
|
||||||
|
return loc
|
||||||
|
}()
|
||||||
|
|||||||
@@ -1847,5 +1847,124 @@ func TestRefreshToken_RevokesOldJTI_DBBacked(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ============================================================
|
||||||
|
// CleanupStaleLoginEntries Tests
|
||||||
|
// ============================================================
|
||||||
|
|
||||||
|
func TestCleanupStaleLoginEntries_Empty(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
loginStateMu.Lock()
|
||||||
|
saved := loginInProgress
|
||||||
|
loginInProgress = make(map[string]time.Time)
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
defer func() {
|
||||||
|
loginStateMu.Lock()
|
||||||
|
loginInProgress = saved
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
|
_, err := CleanupStaleLoginEntries(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanupStaleLoginEntries_RemovesStale(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
loginStateMu.Lock()
|
||||||
|
saved := loginInProgress
|
||||||
|
loginInProgress = map[string]time.Time{
|
||||||
|
"stale-user": clock.Now().Add(-60 * time.Second),
|
||||||
|
}
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
defer func() {
|
||||||
|
loginStateMu.Lock()
|
||||||
|
loginInProgress = saved
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
|
_, err := CleanupStaleLoginEntries(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
loginStateMu.Lock()
|
||||||
|
_, exists := loginInProgress["stale-user"]
|
||||||
|
deleted := loginInProgress["stale-user"]
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
if exists {
|
||||||
|
t.Errorf("expected stale entry (60s old) to be removed, got %v", deleted)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanupStaleLoginEntries_PreservesRecent(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
loginStateMu.Lock()
|
||||||
|
saved := loginInProgress
|
||||||
|
loginInProgress = map[string]time.Time{
|
||||||
|
"recent-user": clock.Now().Add(-5 * time.Second),
|
||||||
|
}
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
defer func() {
|
||||||
|
loginStateMu.Lock()
|
||||||
|
loginInProgress = saved
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
|
_, err := CleanupStaleLoginEntries(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
loginStateMu.Lock()
|
||||||
|
_, exists := loginInProgress["recent-user"]
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
if !exists {
|
||||||
|
t.Error("expected recent entry (5s old) to be preserved")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanupStaleLoginEntries_Mixed(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
loginStateMu.Lock()
|
||||||
|
saved := loginInProgress
|
||||||
|
loginInProgress = map[string]time.Time{
|
||||||
|
"stale-user": clock.Now().Add(-60 * time.Second),
|
||||||
|
"recent-user": clock.Now().Add(-5 * time.Second),
|
||||||
|
"borderline": clock.Now().Add(-29 * time.Second), // Just under 30s threshold
|
||||||
|
}
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
defer func() {
|
||||||
|
loginStateMu.Lock()
|
||||||
|
loginInProgress = saved
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
}()
|
||||||
|
|
||||||
|
_, err := CleanupStaleLoginEntries(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
loginStateMu.Lock()
|
||||||
|
_, staleExists := loginInProgress["stale-user"]
|
||||||
|
_, recentExists := loginInProgress["recent-user"]
|
||||||
|
_, borderlineExists := loginInProgress["borderline"]
|
||||||
|
loginStateMu.Unlock()
|
||||||
|
|
||||||
|
if staleExists {
|
||||||
|
t.Error("expected stale-user (60s old) to be removed")
|
||||||
|
}
|
||||||
|
if !recentExists {
|
||||||
|
t.Error("expected recent-user (5s old) to be preserved")
|
||||||
|
}
|
||||||
|
if !borderlineExists {
|
||||||
|
t.Error("expected borderline entry (29s old) to be preserved")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Ensure test compilation - import pgxpool to avoid unused import
|
// Ensure test compilation - import pgxpool to avoid unused import
|
||||||
var _ = func() *pgxpool.Pool { return nil }
|
var _ = func() *pgxpool.Pool { return nil }
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
package auth
|
package auth
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"crussell/auth"
|
"crussell/auth"
|
||||||
"crussell/clock"
|
"crussell/clock"
|
||||||
"crussell/db"
|
"crussell/db"
|
||||||
"github.com/jackc/pgx/v5"
|
|
||||||
"crussell/internal/dav"
|
"crussell/internal/dav"
|
||||||
"crussell/internal/validators"
|
"crussell/internal/validators"
|
||||||
"crussell/internal/zxcvbnjs"
|
"crussell/internal/zxcvbnjs"
|
||||||
@@ -16,21 +16,22 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
||||||
"github.com/go-chi/chi/v5/middleware"
|
"github.com/jackc/pgx/v5"
|
||||||
|
|
||||||
"os"
|
"os"
|
||||||
"regexp"
|
"regexp"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-chi/chi/v5/middleware"
|
||||||
|
|
||||||
"github.com/nyaruka/phonenumbers"
|
"github.com/nyaruka/phonenumbers"
|
||||||
"golang.org/x/crypto/bcrypt"
|
"golang.org/x/crypto/bcrypt"
|
||||||
"golang.org/x/text/cases"
|
"golang.org/x/text/cases"
|
||||||
"golang.org/x/text/language"
|
"golang.org/x/text/language"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
const maxLoginInProgress = 20
|
const maxLoginInProgress = 20
|
||||||
|
|
||||||
// Login state management
|
// Login state management
|
||||||
@@ -39,30 +40,18 @@ var (
|
|||||||
loginInProgress = make(map[string]time.Time)
|
loginInProgress = make(map[string]time.Time)
|
||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
// CleanupStaleLoginEntries removes stuck loginInProgress entries older than 30 seconds.
|
||||||
go func() {
|
// Called by the centralised jobs scheduler.
|
||||||
ticker := time.NewTicker(1 * time.Hour)
|
func CleanupStaleLoginEntries(ctx context.Context) (int, error) {
|
||||||
defer ticker.Stop()
|
loginStateMu.Lock()
|
||||||
|
defer loginStateMu.Unlock()
|
||||||
for range ticker.C {
|
now := clock.Now()
|
||||||
func() {
|
for userID, startedAt := range loginInProgress {
|
||||||
defer func() {
|
if now.Sub(startedAt) > 30*time.Second {
|
||||||
if r := recover(); r != nil {
|
delete(loginInProgress, userID)
|
||||||
log.Printf("Panic recovered in login state cleanup ticker: %v", r)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
loginStateMu.Lock()
|
|
||||||
now := clock.Now()
|
|
||||||
// Clean up stuck loginInProgress entries (older than 30s)
|
|
||||||
for userID, startedAt := range loginInProgress {
|
|
||||||
if now.Sub(startedAt) > 30*time.Second {
|
|
||||||
delete(loginInProgress, userID)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
loginStateMu.Unlock()
|
|
||||||
}()
|
|
||||||
}
|
}
|
||||||
}()
|
}
|
||||||
|
return 0, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type RegisterRequest struct {
|
type RegisterRequest struct {
|
||||||
|
|||||||
@@ -325,7 +325,7 @@ func TestReserveSlot_DualCleanup(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Run cleanup
|
// Run cleanup
|
||||||
err = scheduling.CleanupOldReservations(ctx)
|
_, err = scheduling.CleanupOldReservations(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("CleanupOldReservations failed: %v", err)
|
t.Fatalf("CleanupOldReservations failed: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -24,28 +24,20 @@ type gdprCacheEntry struct {
|
|||||||
generating bool
|
generating bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func init() {
|
// CleanupGDPRExportCache removes expired entries from the GDPR export cache.
|
||||||
go func() {
|
// Called by the centralised jobs scheduler.
|
||||||
ticker := time.NewTicker(5 * time.Minute)
|
func CleanupGDPRExportCache(ctx context.Context) (int, error) {
|
||||||
defer ticker.Stop()
|
gdprExportCacheMu.Lock()
|
||||||
for range ticker.C {
|
defer gdprExportCacheMu.Unlock()
|
||||||
func() {
|
now := clock.Now()
|
||||||
defer func() {
|
var n int
|
||||||
if r := recover(); r != nil {
|
for k, v := range gdprExportCache {
|
||||||
log.Printf("Panic recovered in GDPR export cache cleanup ticker: %v", r)
|
if now.After(v.expiresAt) {
|
||||||
}
|
delete(gdprExportCache, k)
|
||||||
}()
|
n++
|
||||||
gdprExportCacheMu.Lock()
|
|
||||||
now := clock.Now()
|
|
||||||
for k, v := range gdprExportCache {
|
|
||||||
if now.After(v.expiresAt) {
|
|
||||||
delete(gdprExportCache, k)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
gdprExportCacheMu.Unlock()
|
|
||||||
}()
|
|
||||||
}
|
}
|
||||||
}()
|
}
|
||||||
|
return n, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GET /api/user/gdpr-export
|
// GET /api/user/gdpr-export
|
||||||
|
|||||||
@@ -1214,3 +1214,100 @@ func TestAnonymizeStaleGuestAccounts_DoesNotAffectRegisteredUsers(t *testing.T)
|
|||||||
t.Errorf("expected referral_code to be preserved, got %v", referralCode)
|
t.Errorf("expected referral_code to be preserved, got %v", referralCode)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ============================================================
|
||||||
|
// CleanupGDPRExportCache Tests
|
||||||
|
// ============================================================
|
||||||
|
|
||||||
|
func TestCleanupGDPRExportCache_Empty(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
gdprExportCacheMu.Lock()
|
||||||
|
gdprExportCache = make(map[string]*gdprCacheEntry)
|
||||||
|
gdprExportCacheMu.Unlock()
|
||||||
|
|
||||||
|
_, err := CleanupGDPRExportCache(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanupGDPRExportCache_RemovesExpired(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
gdprExportCacheMu.Lock()
|
||||||
|
gdprExportCache = map[string]*gdprCacheEntry{
|
||||||
|
"user-expired": {expiresAt: clock.Now().Add(-1 * time.Hour)},
|
||||||
|
}
|
||||||
|
gdprExportCacheMu.Unlock()
|
||||||
|
|
||||||
|
_, err := CleanupGDPRExportCache(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
gdprExportCacheMu.RLock()
|
||||||
|
_, exists := gdprExportCache["user-expired"]
|
||||||
|
gdprExportCacheMu.RUnlock()
|
||||||
|
if exists {
|
||||||
|
t.Error("expected expired entry to be removed")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanupGDPRExportCache_PreservesValid(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
gdprExportCacheMu.Lock()
|
||||||
|
gdprExportCache = map[string]*gdprCacheEntry{
|
||||||
|
"user-valid": {expiresAt: clock.Now().Add(1 * time.Hour)},
|
||||||
|
}
|
||||||
|
gdprExportCacheMu.Unlock()
|
||||||
|
|
||||||
|
_, err := CleanupGDPRExportCache(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
gdprExportCacheMu.RLock()
|
||||||
|
entry, exists := gdprExportCache["user-valid"]
|
||||||
|
gdprExportCacheMu.RUnlock()
|
||||||
|
if !exists {
|
||||||
|
t.Fatal("expected valid entry to be preserved")
|
||||||
|
}
|
||||||
|
if entry == nil {
|
||||||
|
t.Error("expected non-nil entry")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCleanupGDPRExportCache_Mixed(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
gdprExportCacheMu.Lock()
|
||||||
|
gdprExportCache = map[string]*gdprCacheEntry{
|
||||||
|
"user-expired": {expiresAt: clock.Now().Add(-2 * time.Hour)},
|
||||||
|
"user-valid": {expiresAt: clock.Now().Add(2 * time.Hour)},
|
||||||
|
"user-expired2": {expiresAt: clock.Now().Add(-30 * time.Minute)},
|
||||||
|
}
|
||||||
|
gdprExportCacheMu.Unlock()
|
||||||
|
|
||||||
|
_, err := CleanupGDPRExportCache(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("expected nil error, got %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
gdprExportCacheMu.RLock()
|
||||||
|
_, expiredExists := gdprExportCache["user-expired"]
|
||||||
|
_, validExists := gdprExportCache["user-valid"]
|
||||||
|
_, expired2Exists := gdprExportCache["user-expired2"]
|
||||||
|
gdprExportCacheMu.RUnlock()
|
||||||
|
|
||||||
|
if expiredExists {
|
||||||
|
t.Error("expected user-expired to be removed")
|
||||||
|
}
|
||||||
|
if expired2Exists {
|
||||||
|
t.Error("expected user-expired2 to be removed")
|
||||||
|
}
|
||||||
|
if !validExists {
|
||||||
|
t.Error("expected user-valid to be preserved")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user