fix expansion issues

Author: rizqme

README.md

@@ -125,6 +125,12 @@ curl -X POST http://localhost:8080/events \
   }'
 ```
 
+> **Note:**
+> - For one-time events, omit the `schedule` field and provide `start_time`.
+> - For recurring events, provide both `start_time` and a `schedule` object.
+> - The `webhook` field is required (not `webhook_url`).
+> - The `Authorization` header is required for all requests.
+
 ## Schedule Parameter Tutorial
 
 The `schedule` parameter in event creation allows you to define recurring or one-time schedules using a flexible JSON structure. Here are the most common use cases:

config.example.yaml

@@ -3,7 +3,7 @@ server:
 
 database:
   host: localhost
-  port: 5432
+  port: 5433
   user: postgres
   password: postgres
   dbname: qhronos
@@ -21,7 +21,8 @@ auth:
 scheduler:
   look_ahead_duration: 24h   # How far into the future to expand recurring events
   expansion_interval: 5m     # How often to run the expander
-  dispatch_worker_count: 4   # Number of dispatch workers
+  dispatch_worker_count: 1   # Number of dispatcher workers
+  grace_period: 2m           # How far into the past to look for missed events
 
 archival:
   check_period: 1h 

docs/api.md

@@ -14,18 +14,34 @@ Authorization: Bearer <token>
 
 ### Create Event
 - **POST** `/events`
-- **Body:**
+- **Body (One-Time Event):**
 ```json
 {
   "name": "My Event",
   "description": "Description...",
   "start_time": "2024-04-23T10:00:00Z",
-  "webhook_url": "https://example.com/webhook",
+  "webhook": "https://example.com/webhook",
   "metadata": {"key": "value"},
+  "tags": ["tag1", "tag2"]
+}
+```
+- **Body (Recurring Event):**
+```json
+{
+  "name": "My Recurring Event",
+  "description": "Description...",
+  "start_time": "2024-04-23T10:00:00Z",
+  "webhook": "https://example.com/webhook",
   "schedule": {"frequency": "daily", "interval": 1},
+  "metadata": {"key": "value"},
   "tags": ["tag1", "tag2"]
 }
 ```
+- **Note:**
+  - For one-time events, omit the `schedule` field and provide `start_time`.
+  - For recurring events, provide both `start_time` and a `schedule` object.
+  - The `webhook` field is required (not `webhook_url`).
+  - The `Authorization` header is required for all requests.
 - **Response:** `201 Created`
 
 ### Get Event

internal/config/config.go

@@ -66,6 +66,7 @@ type SchedulerConfig struct {
 	LookAheadDuration   time.Duration `mapstructure:"look_ahead_duration"`
 	ExpansionInterval   time.Duration `mapstructure:"expansion_interval"`
 	DispatchWorkerCount int           `mapstructure:"dispatch_worker_count"`
+	GracePeriod         time.Duration `mapstructure:"grace_period"`
 }
 
 type RetentionConfig struct {

internal/handlers/event_handler.go

@@ -1,13 +1,15 @@
 package handlers
 
 import (
+	"context"
 	"net/http"
 	"strings"
 	"time"
 
 	"github.com/feedloop/qhronos/internal/middleware"
 	"github.com/feedloop/qhronos/internal/models"
 	"github.com/feedloop/qhronos/internal/repository"
+	"github.com/feedloop/qhronos/internal/scheduler"
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
 	"github.com/lib/pq"
@@ -16,11 +18,12 @@ import (
 )
 
 type EventHandler struct {
-	repo *repository.EventRepository
+	repo     *repository.EventRepository
+	expander *scheduler.Expander
 }
 
-func NewEventHandler(repo *repository.EventRepository) *EventHandler {
-	return &EventHandler{repo: repo}
+func NewEventHandler(repo *repository.EventRepository, expander *scheduler.Expander) *EventHandler {
+	return &EventHandler{repo: repo, expander: expander}
 }
 
 func (h *EventHandler) CreateEvent(c *gin.Context) {
@@ -77,6 +80,21 @@ func (h *EventHandler) CreateEvent(c *gin.Context) {
 		return
 	}
 	logger.Info("Event created", zap.String("event_id", event.ID.String()))
+
+	// Immediate expansion if within window
+	now := time.Now().UTC()
+	graceStart := now.Add(-h.expander.GracePeriod())
+	lookAheadEnd := now.Add(h.expander.LookAheadDuration())
+	if event.StartTime.After(graceStart) && event.StartTime.Before(lookAheadEnd) {
+		go func() {
+			if event.Schedule == nil {
+				_ = h.expander.ExpandNonRecurringEvent(context.Background(), event)
+			} else {
+				_ = h.expander.ExpandRecurringEvent(context.Background(), event)
+			}
+		}()
+	}
+
 	c.JSON(http.StatusCreated, event)
 }
 
@@ -182,17 +200,21 @@ func (h *EventHandler) DeleteEvent(c *gin.Context) {
 func (h *EventHandler) ListEventsByTags(c *gin.Context) {
 	tagsParam := c.Query("tags")
 	if tagsParam == "" {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "tags parameter is required"})
+		// No tags param: return all events
+		events, err := h.repo.List(c)
+		if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		c.JSON(http.StatusOK, events)
 		return
 	}
-
 	tags := strings.Split(tagsParam, ",")
 	events, err := h.repo.ListByTags(c, tags)
 	if err != nil {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
 	}
-
 	c.JSON(http.StatusOK, events)
 }
 

internal/scheduler/dispatcher.go

@@ -183,6 +183,13 @@ func (d *Dispatcher) DispatchWebhook(ctx context.Context, sched *models.Schedule
 	}
 	_ = d.occurrenceRepo.Create(ctx, logOccurrence) // Ignore error to avoid blocking delivery
 
+	// Auto-inactivate one-time events after dispatch (success or max retries)
+	event, err := d.eventRepo.GetByID(ctx, sched.EventID)
+	if err == nil && event != nil && event.Schedule == nil && event.Status == "active" {
+		event.Status = "inactive"
+		_ = d.eventRepo.Update(ctx, event)
+	}
+
 	return err
 }
 

internal/scheduler/expander.go

@@ -17,6 +17,7 @@ type Expander struct {
 	occurrenceRepo    *repository.OccurrenceRepository
 	lookAheadDuration time.Duration
 	expansionInterval time.Duration
+	gracePeriod       time.Duration
 	logger            *zap.Logger
 }
 
@@ -27,6 +28,7 @@ func NewExpander(
 	occurrenceRepo *repository.OccurrenceRepository,
 	lookAheadDuration time.Duration,
 	expansionInterval time.Duration,
+	gracePeriod time.Duration,
 	logger *zap.Logger,
 ) *Expander {
 	logger.Debug("Initializing Expander", zap.Duration("lookAheadDuration", lookAheadDuration), zap.Duration("expansionInterval", expansionInterval))
@@ -36,6 +38,7 @@ func NewExpander(
 		occurrenceRepo:    occurrenceRepo,
 		lookAheadDuration: lookAheadDuration,
 		expansionInterval: expansionInterval,
+		gracePeriod:       gracePeriod,
 		logger:            logger,
 	}
 }
@@ -116,6 +119,7 @@ func (e *Expander) expandRecurringEvent(ctx context.Context, event *models.Event
 
 	// Get next occurrences based on schedule configuration
 	now := time.Now().UTC()
+	graceStart := now.Add(-e.gracePeriod)
 	endTime := now.Add(e.lookAheadDuration)
 	startTime := event.StartTime.UTC()
 
@@ -126,14 +130,14 @@ func (e *Expander) expandRecurringEvent(ctx context.Context, event *models.Event
 	switch schedule.Frequency {
 	case "daily":
 		for t := startTime; t.Before(endTime); t = t.AddDate(0, 0, schedule.Interval) {
-			if t.After(now) {
+			if t.After(graceStart) {
 				occurrences = append(occurrences, t)
 			}
 		}
 	case "weekly":
 		if len(schedule.ByDay) == 0 {
 			for t := startTime; t.Before(endTime); t = t.AddDate(0, 0, 7*schedule.Interval) {
-				if t.After(now) {
+				if t.After(graceStart) {
 					occurrences = append(occurrences, t)
 				}
 			}
@@ -152,7 +156,7 @@ func (e *Expander) expandRecurringEvent(ctx context.Context, event *models.Event
 					weekday := weekdayMap[day]
 					daysToAdd := (int(weekday) - int(t.Weekday()) + 7) % 7
 					nextDay := t.AddDate(0, 0, daysToAdd)
-					if nextDay.After(now) && nextDay.Before(endTime) {
+					if nextDay.After(graceStart) && nextDay.Before(endTime) {
 						occurrences = append(occurrences, nextDay)
 					}
 				}
@@ -161,15 +165,15 @@ func (e *Expander) expandRecurringEvent(ctx context.Context, event *models.Event
 	case "monthly":
 		if len(schedule.ByMonthDay) == 0 {
 			for t := startTime; t.Before(endTime); t = t.AddDate(0, schedule.Interval, 0) {
-				if t.After(now) {
+				if t.After(graceStart) {
 					occurrences = append(occurrences, t)
 				}
 			}
 		} else {
 			for t := startTime; t.Before(endTime); t = t.AddDate(0, schedule.Interval, 0) {
 				for _, day := range schedule.ByMonthDay {
 					nextDay := time.Date(t.Year(), t.Month(), day, t.Hour(), t.Minute(), t.Second(), t.Nanosecond(), t.Location())
-					if nextDay.After(now) && nextDay.Before(endTime) {
+					if nextDay.After(graceStart) && nextDay.Before(endTime) {
 						occurrences = append(occurrences, nextDay)
 					}
 				}
@@ -178,15 +182,15 @@ func (e *Expander) expandRecurringEvent(ctx context.Context, event *models.Event
 	case "yearly":
 		if len(schedule.ByMonth) == 0 {
 			for t := startTime; t.Before(endTime); t = t.AddDate(schedule.Interval, 0, 0) {
-				if t.After(now) {
+				if t.After(graceStart) {
 					occurrences = append(occurrences, t)
 				}
 			}
 		} else {
 			for t := startTime; t.Before(endTime); t = t.AddDate(schedule.Interval, 0, 0) {
 				for _, month := range schedule.ByMonth {
 					nextMonth := time.Date(t.Year(), time.Month(month), t.Day(), t.Hour(), t.Minute(), t.Second(), t.Nanosecond(), t.Location())
-					if nextMonth.After(now) && nextMonth.Before(endTime) {
+					if nextMonth.After(graceStart) && nextMonth.Before(endTime) {
 						occurrences = append(occurrences, nextMonth)
 					}
 				}
@@ -239,13 +243,18 @@ func (e *Expander) expandNonRecurringEvent(ctx context.Context, event *models.Ev
 	}
 
 	now := time.Now().UTC()
+	graceStart := now.Add(-e.gracePeriod)
 	startTime := event.StartTime.UTC()
 
 	endTime := now.Add(e.lookAheadDuration)
 	if startTime.After(endTime) {
 		e.logger.Info("Event is beyond look-ahead window, skipping", zap.String("event_id", event.ID.String()))
 		return nil
 	}
+	if startTime.Before(graceStart) {
+		e.logger.Info("Event is before grace period, skipping", zap.String("event_id", event.ID.String()))
+		return nil
+	}
 
 	occurrence := &models.Occurrence{
 		OccurrenceID: uuid.New(),
@@ -277,3 +286,19 @@ func (e *Expander) Run(ctx context.Context) error {
 		}
 	}
 }
+
+func (e *Expander) ExpandRecurringEvent(ctx context.Context, event *models.Event) error {
+	return e.expandRecurringEvent(ctx, event)
+}
+
+func (e *Expander) ExpandNonRecurringEvent(ctx context.Context, event *models.Event) error {
+	return e.expandNonRecurringEvent(ctx, event)
+}
+
+func (e *Expander) GracePeriod() time.Duration {
+	return e.gracePeriod
+}
+
+func (e *Expander) LookAheadDuration() time.Duration {
+	return e.lookAheadDuration
+}

internal/scheduler/expander_test.go

@@ -38,8 +38,9 @@ func TestEventExpander(t *testing.T) {
 	// Test configuration
 	lookAheadDuration := 24 * time.Hour
 	expansionInterval := 5 * time.Minute
+	gracePeriod := 2 * time.Minute
 
-	expander := NewExpander(scheduler, eventRepo, occurrenceRepo, lookAheadDuration, expansionInterval, logger)
+	expander := NewExpander(scheduler, eventRepo, occurrenceRepo, lookAheadDuration, expansionInterval, gracePeriod, logger)
 
 	t.Run("successful event expansion", func(t *testing.T) {
 		cleanup()

main.go

@@ -187,6 +187,7 @@ func main() {
 		occurrenceRepo,
 		cfg.Scheduler.LookAheadDuration,
 		cfg.Scheduler.ExpansionInterval,
+		cfg.Scheduler.GracePeriod,
 		logger,
 	)
 
@@ -196,7 +197,7 @@ func main() {
 	dispatcher := scheduler.NewDispatcher(eventRepo, occurrenceRepo, hmacService, logger)
 
 	// Initialize handlers
-	eventHandler := handlers.NewEventHandler(eventRepo)
+	eventHandler := handlers.NewEventHandler(eventRepo, expander)
 	occurrenceHandler := handlers.NewOccurrenceHandler(eventRepo, occurrenceRepo)
 	tokenHandler := handlers.NewTokenHandler(tokenService)