diff --git a/README.MD b/README.MD index 846ad1c..3a4262a 100644 --- a/README.MD +++ b/README.MD @@ -65,7 +65,7 @@ That gives you intelligent routing without putting an LLM call in the `Publish() - **Dead-letter queue** with replay support and panic recovery. - **Redis Streams and RabbitMQ EventStore options** for cross-process delivery. - **Transactions** with local buffering, synchronous commit, and optional Rules/Cache/Jev event selection. -- **Scheduling** with absolute and relative timers. +- **Scheduling** with absolute/relative timers and optional Rules/Cache/Jev event selection. - **Metrics** for published, processed, and failed events. - **Pluggable decision cache** through the `DecisionCache` interface. @@ -322,6 +322,7 @@ The repository includes runnable examples for both the classic event-bus path an | [`routing_cache`](examples/routing_cache) | Reusing a previous decision so the fallback selector is called only once | | [`routing_jev`](examples/routing_jev) | Full `rules → cache → Jev → GoEventBus` routing with OpenRouter | | [`transaction_jev`](examples/transaction_jev) | Rules/Cache/Jev event selection buffered until transaction commit | +| [`schedule_jev`](examples/schedule_jev) | Rules/Cache/Jev event selection before delayed scheduling | | [`hello_world`](examples/hello_world) | Minimal direct `Subscribe → Publish` flow | | [`middleware`](examples/middleware) | Handler middleware and lifecycle behavior | | [`goroutines-subscribe-publisher`](examples/goroutines-subscribe-publisher) | Concurrent producers and publishing | @@ -340,6 +341,7 @@ go run ./examples/routing_rules go run ./examples/routing_cache OPENROUTER_API_KEY=... go run ./examples/routing_jev OPENROUTER_API_KEY=... go run ./examples/transaction_jev +OPENROUTER_API_KEY=... go run ./examples/schedule_jev ``` See [`examples/README.md`](examples/README.md) for a compact guide to all examples. @@ -870,7 +872,7 @@ shared EventStore ring. # Scheduling -Schedule an event at a time: +Schedule a known event at a specific time: ```go timer := store.Schedule( @@ -890,9 +892,85 @@ timer := store.ScheduleAfter( ) ``` -The returned `*time.Timer` may be stopped before it fires. +## Intelligent scheduling -Past times and non-positive durations are handled immediately. +Use the same Rules / Cache / Jev selector pipeline when the event type must be +chosen before it is scheduled. + +Schedule at an absolute time: + +```go +decision, timer, err := store.DecideAndSchedule( + ctx, + time.Now().Add(10*time.Second), + selector, + state, + GoEventBus.Event{ + ID: "evt-scheduled-1", + Data: payload, + }, + candidates, +) +if err != nil { + log.Fatal(err) +} + +fmt.Println("scheduled:", decision.Choice) +``` + +Or after a duration: + +```go +decision, timer, err := store.DecideAndScheduleAfter( + ctx, + 30*time.Second, + selector, + state, + GoEventBus.Event{ + ID: "evt-scheduled-2", + Data: payload, + }, + candidates, +) +if err != nil { + log.Fatal(err) +} +``` + +The selector is evaluated **when scheduling is requested**, not when the timer +fires: + +```text +state + ↓ +Rules + ↓ +Cache + ↓ +Jev + ↓ +selected event + ↓ +timer + ↓ +Subscribe → Publish → handler +``` + +That means selection errors are returned immediately and no timer is created. +Once scheduling succeeds, the timer contains an already-resolved event, so a +later change in rules, cache contents, or Jev output does not change what will +fire. + +`DecideAndSchedule` and `DecideAndScheduleAfter` work with +`RuleCacheSelector`, `JevSelector`, or any custom `EventSelector`. + +The returned `*time.Timer` may be stopped before it fires. Past times and +non-positive durations preserve the existing immediate-execution behavior and +return a nil timer. + +Scheduling remains a **local EventStore operation**. It does not publish the +scheduled event to Redis/RabbitMQ even when the parent store has a broker +configured. --- @@ -975,7 +1053,8 @@ Important methods: | `Use` | Register middleware | | `OnBefore` / `OnAfter` / `OnError` | Register lifecycle hooks | | `Metrics` | Read event counters | -| `Schedule` / `ScheduleAfter` | Schedule future events | +| `Schedule` / `ScheduleAfter` | Schedule known events | +| `DecideAndSchedule` / `DecideAndScheduleAfter` | Select with Rules/Cache/Jev, then schedule the resolved event | | `BeginTransaction` | Start a local transaction buffer; use `tx.DecideAndPublish` for Rules/Cache/Jev selection | | `Drain` / `Close` | Stop accepting events and finish in-flight work | diff --git a/examples/README.md b/examples/README.md index 7bb9784..7f0cf43 100644 --- a/examples/README.md +++ b/examples/README.md @@ -73,6 +73,22 @@ Flow: state -> rules/cache/Jev -> selected event -> transaction buffer -> Commit -> handler ``` +### `schedule_jev` + +Uses Rules / Cache / Jev to resolve the event type before starting a timer. +The decision is made immediately; the selected event is dispatched only when +the timer fires. + +```bash +OPENROUTER_API_KEY=... go run ./examples/schedule_jev +``` + +Flow: + +```text +state -> rules/cache/Jev -> selected event -> timer -> Subscribe -> Publish -> handler +``` + ## Core event bus | Example | Purpose | @@ -89,6 +105,7 @@ state -> rules/cache/Jev -> selected event -> transaction buffer -> Commit -> ha | `redis` | Redis Streams configured directly on `NewEventStore` | | `rabbitmq` | RabbitMQ configured directly on `NewEventStore` | | `transaction_jev` | Rules/Cache/Jev event selection buffered until transaction commit | +| `schedule_jev` | Rules/Cache/Jev event selection before delayed scheduling | ## Broker-backed stores diff --git a/examples/schedule_jev/main.go b/examples/schedule_jev/main.go new file mode 100644 index 0000000..8f904e6 --- /dev/null +++ b/examples/schedule_jev/main.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "fmt" + "log" + "os" + "time" + + GoEventBus "github.com/Protocol-Lattice/GoEventBus" +) + +func main() { + apiKey := os.Getenv("OPENROUTER_API_KEY") + if apiKey == "" { + fmt.Println("set OPENROUTER_API_KEY to run this example") + return + } + + ctx := context.Background() + dispatcher := GoEventBus.Dispatcher{} + dispatcher.Register("order.created", printHandler) + dispatcher.Register("invoice.created", printHandler) + + store := GoEventBus.NewEventStore(&dispatcher, 16, GoEventBus.DropOldest) + defer func() { _ = store.Close(context.Background()) }() + + selector := &GoEventBus.RuleCacheSelector{ + Cache: GoEventBus.NewMemoryDecisionCache(5 * time.Minute), + Fallback: &GoEventBus.JevSelector{ + APIKey: apiKey, + }, + } + + candidates := []GoEventBus.EventCandidate{ + { + Key: "order_created", + Projection: "order.created", + Description: "A new order should be created", + }, + { + Key: "invoice_created", + Projection: "invoice.created", + Description: "An invoice should be created for an existing order", + }, + } + + decision, timer, err := store.DecideAndScheduleAfter( + ctx, + 2*time.Second, + selector, + map[string]any{ + "message": "Create an invoice for order 42 in two seconds", + }, + GoEventBus.Event{ + ID: "evt-scheduled-1", + Data: map[string]any{"order_id": "42"}, + }, + candidates, + ) + if err != nil { + log.Fatal(err) + } + + fmt.Printf( + "scheduled choice=%s confidence=%.2f\n", + decision.Choice, + decision.Confidence, + ) + + if timer == nil { + return + } + + select { + case <-time.After(3 * time.Second): + case <-ctx.Done(): + log.Fatal(ctx.Err()) + } +} + +func printHandler( + _ context.Context, + ev GoEventBus.Event, +) (GoEventBus.Result, error) { + fmt.Printf("handled projection=%v event=%s\n", ev.Projection, ev.ID) + return GoEventBus.Result{Message: "ok"}, nil +} diff --git a/jev.go b/jev.go index 7749239..7162385 100644 --- a/jev.go +++ b/jev.go @@ -13,8 +13,8 @@ import ( ) var ( - // ErrNilEventSelector is returned when DecideAndSubscribe is called without - // an EventSelector. + // ErrNilEventSelector is returned when a selector-aware operation is called + // without an EventSelector. ErrNilEventSelector = errors.New("goeventbus: event selector is nil") // ErrNoEventCandidates is returned when there are no event types to choose from. ErrNoEventCandidates = errors.New("goeventbus: no event candidates") diff --git a/scheduling_jev.go b/scheduling_jev.go new file mode 100644 index 0000000..f67ec95 --- /dev/null +++ b/scheduling_jev.go @@ -0,0 +1,51 @@ +package GoEventBus + +import ( + "context" + "time" +) + +// DecideAndSchedule asks selector to choose an event type immediately, assigns +// the chosen projection, and schedules the resolved event for t. +// +// Selection happens when DecideAndSchedule is called, not when the timer fires. +// This makes routing failures visible immediately and guarantees that the timer +// always represents one already-resolved event. The scheduled event follows the +// existing local Schedule semantics. +func (es *EventStore) DecideAndSchedule( + ctx context.Context, + t time.Time, + selector EventSelector, + state any, + event Event, + candidates []EventCandidate, +) (EventDecision, *time.Timer, error) { + decision, selected, err := selectEvent(ctx, selector, state, event, candidates) + if err != nil { + return EventDecision{}, nil, err + } + + return decision, es.Schedule(ctx, t, selected), nil +} + +// DecideAndScheduleAfter asks selector to choose an event type immediately, +// assigns the chosen projection, and schedules the resolved event after d. +// +// Selection happens when DecideAndScheduleAfter is called. A non-positive +// duration preserves ScheduleAfter's immediate-execution behavior and returns +// a nil timer. +func (es *EventStore) DecideAndScheduleAfter( + ctx context.Context, + d time.Duration, + selector EventSelector, + state any, + event Event, + candidates []EventCandidate, +) (EventDecision, *time.Timer, error) { + decision, selected, err := selectEvent(ctx, selector, state, event, candidates) + if err != nil { + return EventDecision{}, nil, err + } + + return decision, es.ScheduleAfter(ctx, d, selected), nil +} diff --git a/scheduling_jev_test.go b/scheduling_jev_test.go new file mode 100644 index 0000000..1068835 --- /dev/null +++ b/scheduling_jev_test.go @@ -0,0 +1,212 @@ +package GoEventBus + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func TestDecideAndScheduleAfterWithJev(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("method = %s, want POST", r.Method) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "model":"typesafe/jev-1.13", + "id":"schedule-jev-test", + "answers":{ + "event_type":{ + "type":"choice", + "choice":"invoice_created", + "probabilities":{ + "invoice_created":0.97, + "order_created":0.03 + }, + "confidence":0.94 + } + } + }`)) + })) + defer server.Close() + + delivered := make(chan Event, 1) + dispatcher := Dispatcher{} + dispatcher.Register("invoice.created", func(_ context.Context, event Event) (Result, error) { + delivered <- event + return Result{Message: "ok"}, nil + }) + + store := NewEventStore(&dispatcher, 8, DropOldest) + t.Cleanup(func() { _ = store.Close(context.Background()) }) + + selector := &JevSelector{ + APIKey: "test-key", + Endpoint: server.URL, + Client: server.Client(), + } + + decision, timer, err := store.DecideAndScheduleAfter( + context.Background(), + 75*time.Millisecond, + selector, + map[string]any{"message": "create an invoice for order 42"}, + Event{ID: "scheduled-1", Data: map[string]any{"order_id": "42"}}, + []EventCandidate{ + { + Key: "order_created", + Projection: "order.created", + Description: "A new order should be created", + }, + { + Key: "invoice_created", + Projection: "invoice.created", + Description: "An invoice should be created", + }, + }, + ) + if err != nil { + t.Fatalf("DecideAndScheduleAfter: %v", err) + } + if timer == nil { + t.Fatal("timer = nil, want scheduled timer") + } + if decision.Choice != "invoice_created" { + t.Fatalf("choice = %q, want invoice_created", decision.Choice) + } + if decision.Projection != "invoice.created" { + t.Fatalf("projection = %#v, want invoice.created", decision.Projection) + } + if decision.RequestID != "schedule-jev-test" { + t.Fatalf("request ID = %q, want schedule-jev-test", decision.RequestID) + } + + select { + case event := <-delivered: + t.Fatalf("event fired before delay: %#v", event) + case <-time.After(20 * time.Millisecond): + } + + select { + case event := <-delivered: + if event.ID != "scheduled-1" { + t.Fatalf("event ID = %q, want scheduled-1", event.ID) + } + if event.Projection != "invoice.created" { + t.Fatalf("event projection = %#v, want invoice.created", event.Projection) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for scheduled event") + } +} + +func TestDecideAndSchedulePastTimeExecutesSelectedEventImmediately(t *testing.T) { + var processed uint64 + dispatcher := Dispatcher{} + dispatcher.Register("selected", func(_ context.Context, event Event) (Result, error) { + if event.ID != "immediate-1" { + t.Fatalf("event ID = %q, want immediate-1", event.ID) + } + atomic.AddUint64(&processed, 1) + return Result{}, nil + }) + + store := NewEventStore(&dispatcher, 8, DropOldest) + t.Cleanup(func() { _ = store.Close(context.Background()) }) + + selector := eventSelectorFunc(func( + context.Context, + any, + []EventCandidate, + ) (EventDecision, error) { + return EventDecision{Choice: "selected", Confidence: 1}, nil + }) + + decision, timer, err := store.DecideAndSchedule( + context.Background(), + time.Now().Add(-time.Second), + selector, + "state", + Event{ID: "immediate-1"}, + []EventCandidate{{Key: "selected", Projection: "selected"}}, + ) + if err != nil { + t.Fatalf("DecideAndSchedule: %v", err) + } + if timer != nil { + t.Fatal("timer != nil for past schedule") + } + if decision.Projection != "selected" { + t.Fatalf("projection = %#v, want selected", decision.Projection) + } + if got := atomic.LoadUint64(&processed); got != 1 { + t.Fatalf("processed = %d, want 1", got) + } +} + +func TestDecideAndScheduleAfterNonPositiveExecutesImmediately(t *testing.T) { + var processed uint64 + dispatcher := Dispatcher{ + "selected": hs(makeCounterHandler(&processed)), + } + store := NewEventStore(&dispatcher, 8, DropOldest) + t.Cleanup(func() { _ = store.Close(context.Background()) }) + + selector := eventSelectorFunc(func( + context.Context, + any, + []EventCandidate, + ) (EventDecision, error) { + return EventDecision{Choice: "selected"}, nil + }) + + _, timer, err := store.DecideAndScheduleAfter( + context.Background(), + 0, + selector, + nil, + Event{ID: "immediate-after"}, + []EventCandidate{{Key: "selected", Projection: "selected"}}, + ) + if err != nil { + t.Fatalf("DecideAndScheduleAfter: %v", err) + } + if timer != nil { + t.Fatal("timer != nil for non-positive duration") + } + if got := atomic.LoadUint64(&processed); got != 1 { + t.Fatalf("processed = %d, want 1", got) + } +} + +func TestDecideAndScheduleSelectionFailureCreatesNoTimer(t *testing.T) { + store := NewEventStore(&Dispatcher{}, 8, DropOldest) + t.Cleanup(func() { _ = store.Close(context.Background()) }) + + selector := eventSelectorFunc(func( + context.Context, + any, + []EventCandidate, + ) (EventDecision, error) { + return EventDecision{Choice: "unknown"}, nil + }) + + _, timer, err := store.DecideAndScheduleAfter( + context.Background(), + time.Hour, + selector, + "state", + Event{ID: "never-scheduled"}, + []EventCandidate{{Key: "known", Projection: "known"}}, + ) + if !errors.Is(err, ErrUnknownEventChoice) { + t.Fatalf("error = %v, want ErrUnknownEventChoice", err) + } + if timer != nil { + t.Fatal("timer created for failed selection") + } +} diff --git a/website/dist/docs/index.html b/website/dist/docs/index.html index 37ead82..aaec243 100644 --- a/website/dist/docs/index.html +++ b/website/dist/docs/index.html @@ -679,7 +679,8 @@

Transactions

Operations

Scheduling#

-

Schedule an event at a time:

+

Direct scheduling#

+

Schedule a known event at a specific time:

Go
timer := store.Schedule(
     ctx,
     time.Now().Add(10*time.Second),
@@ -691,9 +692,45 @@ 
Scheduling
-

The returned *time.Timer may be stopped before it fires.

-

Past times and non-positive durations are handled immediately.

- +
+

Jev-aware scheduling#

+

Use the same Rules / Cache / Jev selector pipeline when the event type must be chosen before scheduling.

+
Go
decision, timer, err := store.DecideAndSchedule(
+    ctx,
+    time.Now().Add(10*time.Second),
+    selector,
+    state,
+    GoEventBus.Event{
+        ID:   "evt-scheduled-1",
+        Data: payload,
+    },
+    candidates,
+)
+if err != nil {
+    log.Fatal(err)
+}
+
+fmt.Println("scheduled:", decision.Choice)
+

Or after a duration:

+
Go
decision, timer, err := store.DecideAndScheduleAfter(
+    ctx,
+    30*time.Second,
+    selector,
+    state,
+    GoEventBus.Event{
+        ID:   "evt-scheduled-2",
+        Data: payload,
+    },
+    candidates,
+)
+if err != nil {
+    log.Fatal(err)
+}
+
01Rules / Cache / Jev02Selected event03Timer04Handler
+

The selector runs when scheduling is requested, not when the timer fires. Selection errors are returned immediately and no timer is created. A successfully scheduled timer therefore contains an already-resolved event.

+ +
+

The returned *time.Timer may be stopped before it fires. Past times and non-positive durations execute immediately and return nil.

Operations

@@ -762,7 +799,8 @@

API referenceUseRegister middleware OnBefore / OnAfter / OnErrorRegister lifecycle hooks MetricsRead event counters -Schedule / ScheduleAfterSchedule future events +Schedule / ScheduleAfterSchedule known events +DecideAndSchedule / DecideAndScheduleAfterSelect with Rules/Cache/Jev, then schedule the resolved event BeginTransactionStart a local transaction buffer; use tx.DecideAndPublish for Rules/Cache/Jev selection Drain / CloseStop accepting events and finish in-flight work @@ -777,6 +815,7 @@

Examplesrouting_cacheReusing a previous decision so the fallback selector is called only once routing_jevFull rules → cache → Jev → GoEventBus routing with OpenRouter transaction_jevRules/Cache/Jev event selection buffered until transaction commit +schedule_jevRules/Cache/Jev event selection before delayed scheduling hello_worldMinimal direct Subscribe → Publish flow middlewareHandler middleware and lifecycle behavior goroutines-subscribe-publisherConcurrent producers and publishing @@ -790,7 +829,8 @@

Examples
Terminal
go run ./examples/routing_rules
 go run ./examples/routing_cache
 OPENROUTER_API_KEY=... go run ./examples/routing_jev
-OPENROUTER_API_KEY=... go run ./examples/transaction_jev
+OPENROUTER_API_KEY=... go run ./examples/transaction_jev +OPENROUTER_API_KEY=... go run ./examples/schedule_jev

See examples/README.md for a compact guide to all examples.