Add scheduler package

Author: adrianpk

docs/features.md

@@ -55,6 +55,7 @@
 |---------|-------------|
 | `mailer` | Email delivery (SMTP, SendGrid, AWS SES) |
 | `pubsub` | Publish/subscribe messaging |
+| `scheduler` | Job scheduling with pluggable storage, concurrent workers, dynamic settings |
 | `telemetry` | Request counting, crash collection, settings schemas |
 | `testhelper` | Testing utilities |
 | `fake` | Test doubles (mailer, telemetry) |

scheduler/README.md

@@ -0,0 +1,128 @@
+# scheduler
+
+Job scheduler with pluggable storage backends.
+
+## Features
+
+- **Pluggable storage**: Bring your own `JobStore` implementation (Postgres included)
+- **Concurrent workers**: Configurable worker pool for parallel job execution
+- **Dynamic configuration**: Override settings at runtime via `SettingsProvider`
+- **Testable**: Fake clock, store, and logger for deterministic tests
+- **Schedule types**: Daily, Weekly, and Interval schedules with timezone support
+
+## Usage
+
+```go
+package main
+
+import (
+    "context"
+    "log"
+    "time"
+
+    "github.com/hatmaxkit/hatmax/scheduler"
+    "github.com/hatmaxkit/hatmax/scheduler/postgres"
+)
+
+func main() {
+    db := connectDB()
+    store := postgres.NewStore(db)
+
+    cfg := scheduler.Config{
+        Enabled:   true,
+        Interval:  30 * time.Second,
+        BatchSize: 20,
+        Workers:   4,
+    }
+
+    sched := scheduler.New(store, cfg, logger)
+
+    sched.Register("send-email", func(ctx context.Context, job scheduler.Job) scheduler.Result {
+        // Process job
+        return scheduler.Result{Output: map[string]any{"sent": true}}
+    })
+
+    sched.Start(ctx)
+    defer sched.Stop(ctx)
+}
+```
+
+## Configuration
+
+### Static (Config struct)
+
+| Field | Default | Description |
+|-------|---------|-------------|
+| Enabled | false | Enable/disable scheduler |
+| Interval | 1m | Polling interval |
+| BatchSize | 20 | Max jobs per tick |
+| Workers | 1 | Concurrent workers |
+| RetryAttempts | 3 | Max retry attempts |
+| RetryBackoff | 1m | Base backoff duration |
+
+### Dynamic (SettingsProvider)
+
+Implement `SettingsProvider` to override at runtime:
+
+```go
+sched.SetSettings(settingsService)
+```
+
+Settings keys:
+- `scheduler.enabled` - Override Enabled
+- `scheduler.interval_seconds` - Override Interval
+- `scheduler.paused` - Pause without stopping
+
+## Schedule Types
+
+```go
+// Daily at 9:00 AM UTC
+daily := scheduler.Daily{Hour: 9, Minute: 0}
+
+// Weekly on Friday at 5:00 PM in New York
+loc, _ := time.LoadLocation("America/New_York")
+weekly := scheduler.Weekly{Day: time.Friday, Hour: 17, Minute: 0, TZ: loc}
+
+// Every 30 minutes
+interval := scheduler.Interval{Every: 30 * time.Minute}
+
+// Get next run time
+next := daily.Next(time.Now())
+```
+
+## Testing
+
+Use fakes for deterministic tests:
+
+```go
+store := scheduler.NewFakeStore()
+clock := scheduler.NewFakeClock(baseTime)
+log := &scheduler.FakeLogger{}
+
+sched := scheduler.New(store, cfg, log)
+sched.SetClock(clock)
+
+store.AddJob(scheduler.Job{
+    ID:           "test-job",
+    TaskType:     "email",
+    ScheduledFor: clock.Now(),
+})
+
+sched.Tick(ctx)
+clock.Advance(time.Hour)
+```
+
+## Postgres Backend
+
+```go
+import "github.com/hatmaxkit/hatmax/scheduler/postgres"
+
+store := postgres.NewStore(db)
+
+// Apply schema (or use migrations)
+db.Exec(postgres.Schema)
+```
+
+Tables:
+- `scheduled_jobs` - Job definitions
+- `job_runs` - Execution history

scheduler/config.go

@@ -0,0 +1,37 @@
+package scheduler
+
+import "time"
+
+type Config struct {
+	Enabled       bool
+	Interval      time.Duration
+	BatchSize     int
+	Workers       int
+	RetryAttempts int
+	RetryBackoff  time.Duration
+}
+
+func (c Config) WithDefaults() Config {
+	if c.Interval <= 0 {
+		c.Interval = time.Minute
+	}
+	if c.BatchSize <= 0 {
+		c.BatchSize = 20
+	}
+	if c.Workers <= 0 {
+		c.Workers = 1
+	}
+	if c.RetryAttempts <= 0 {
+		c.RetryAttempts = 3
+	}
+	if c.RetryBackoff <= 0 {
+		c.RetryBackoff = time.Minute
+	}
+	return c
+}
+
+const (
+	SettingEnabled  = "scheduler.enabled"
+	SettingInterval = "scheduler.interval_seconds"
+	SettingPaused   = "scheduler.paused"
+)

scheduler/fake.go

@@ -0,0 +1,161 @@
+package scheduler
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+type FakeClock struct {
+	mu  sync.Mutex
+	now time.Time
+}
+
+func NewFakeClock(t time.Time) *FakeClock {
+	return &FakeClock{now: t}
+}
+
+func (c *FakeClock) Now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.now
+}
+
+func (c *FakeClock) Set(t time.Time) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.now = t
+}
+
+func (c *FakeClock) Advance(d time.Duration) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.now = c.now.Add(d)
+}
+
+type FakeStore struct {
+	mu   sync.Mutex
+	Jobs []Job
+	Runs map[string]*FakeRun
+}
+
+type FakeRun struct {
+	ID           string
+	JobID        string
+	ScheduledFor time.Time
+	StartedAt    time.Time
+	FinishedAt   time.Time
+	Status       string
+	Output       []byte
+	Error        string
+}
+
+func NewFakeStore() *FakeStore {
+	return &FakeStore{
+		Runs: make(map[string]*FakeRun),
+	}
+}
+
+func (s *FakeStore) AddJob(job Job) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.Jobs = append(s.Jobs, job)
+}
+
+func (s *FakeStore) ListDue(ctx context.Context, now time.Time, limit int) ([]Job, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var due []Job
+	for _, j := range s.Jobs {
+		if !j.ScheduledFor.After(now) {
+			due = append(due, j)
+			if len(due) >= limit {
+				break
+			}
+		}
+	}
+	return due, nil
+}
+
+func (s *FakeStore) CreateRun(ctx context.Context, jobID, runID string, scheduledFor time.Time) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.Runs[runID] = &FakeRun{
+		ID:           runID,
+		JobID:        jobID,
+		ScheduledFor: scheduledFor,
+		Status:       "pending",
+	}
+	return nil
+}
+
+func (s *FakeStore) MarkRunning(ctx context.Context, runID string, startedAt time.Time) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if run, ok := s.Runs[runID]; ok {
+		run.Status = "running"
+		run.StartedAt = startedAt
+	}
+	return nil
+}
+
+func (s *FakeStore) MarkSuccess(ctx context.Context, runID string, finishedAt time.Time, output []byte) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if run, ok := s.Runs[runID]; ok {
+		run.Status = "success"
+		run.FinishedAt = finishedAt
+		run.Output = output
+	}
+	return nil
+}
+
+func (s *FakeStore) MarkFailed(ctx context.Context, runID string, finishedAt time.Time, errMsg string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if run, ok := s.Runs[runID]; ok {
+		run.Status = "failed"
+		run.FinishedAt = finishedAt
+		run.Error = errMsg
+	}
+	return nil
+}
+
+func (s *FakeStore) UpdateNextRun(ctx context.Context, jobID string, lastRun, nextRun time.Time) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for i, j := range s.Jobs {
+		if j.ID == jobID {
+			s.Jobs[i].ScheduledFor = nextRun
+			break
+		}
+	}
+	return nil
+}
+
+func (s *FakeStore) GetRun(runID string) *FakeRun {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.Runs[runID]
+}
+
+func (s *FakeStore) AllRuns() []*FakeRun {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	runs := make([]*FakeRun, 0, len(s.Runs))
+	for _, r := range s.Runs {
+		runs = append(runs, r)
+	}
+	return runs
+}
+
+type FakeLogger struct {
+	mu       sync.Mutex
+	Messages []string
+}
+
+func (l *FakeLogger) Info(v ...any)                  {}
+func (l *FakeLogger) Infof(format string, a ...any)  {}
+func (l *FakeLogger) Error(v ...any)                 {}
+func (l *FakeLogger) Errorf(format string, a ...any) {}