store/jetstream.go

@@ -191,29 +191,19 @@ func (jes *JetStreamEventStore) SaveEvent(event *aether.Event) error {
 
 // getLatestVersionLocked returns the latest version for an actor.
 // Caller must hold jes.mu.
+// This method uses the optimized GetLatestVersion which fetches only the last message.
 func (jes *JetStreamEventStore) getLatestVersionLocked(actorID string) (int64, error) {
 	// Check cache first
 	if version, ok := jes.versions[actorID]; ok {
 		return version, nil
 	}
 
-	// Fetch from JetStream - use internal method that returns result
+	// Use optimized GetLatestVersion to fetch only last event
-	result, err := jes.getEventsWithErrorsInternal(actorID, 0)
+	latestVersion, err := jes.GetLatestVersion(actorID)
 	if err != nil {
 		return 0, err
 	}
 
-	if len(result.Events) == 0 {
-		return 0, nil
-	}
-
-	latestVersion := int64(0)
-	for _, event := range result.Events {
-		if event.Version > latestVersion {
-			latestVersion = event.Version
-		}
-	}
-
 	// Update cache
 	jes.versions[actorID] = latestVersion
 
@@ -303,38 +293,46 @@ func (jes *JetStreamEventStore) getEventsWithErrorsInternal(actorID string, from
 	return result, nil
 }
 
-// GetLatestVersion returns the latest version for an actor, repopulating cache
+// GetLatestVersion returns the latest version for an actor in O(1) time.
-// with fresh data to ensure consistency even if external processes write to
+// It uses JetStream's DeliverLast() option to fetch only the last message
-// the same JetStream stream.
+// instead of scanning all events, making this O(1) instead of O(n).
 func (jes *JetStreamEventStore) GetLatestVersion(actorID string) (int64, error) {
-	// Hold lock during fetch to prevent race condition with SaveEvent
+	// Create subject filter for this actor
-	jes.mu.Lock()
+	subject := fmt.Sprintf("%s.events.%s.%s",
-	defer jes.mu.Unlock()
+		jes.streamName,
+		sanitizeSubject(extractActorType(actorID)),
+		sanitizeSubject(actorID))
 
-	events, err := jes.GetEvents(actorID, 0)
+	// Create consumer to read only the last message
+	consumer, err := jes.js.PullSubscribe(subject, "", nats.DeliverLast())
 	if err != nil {
-		return 0, err
+		return 0, fmt.Errorf("failed to create consumer: %w", err)
+	}
+	defer consumer.Unsubscribe()
+
+	// Fetch only the last message
+	msgs, err := consumer.Fetch(1, nats.MaxWait(time.Second))
+	if err != nil {
+		if err == nats.ErrTimeout {
+			// No messages for this actor, return 0
+			return 0, nil
+		}
+		return 0, fmt.Errorf("failed to fetch last message: %w", err)
 	}
 
-	if len(events) == 0 {
+	if len(msgs) == 0 {
-		// No events for this actor - ensure cache is cleared
+		// No events for this actor
-		delete(jes.versions, actorID)
 		return 0, nil
 	}
 
-	latestVersion := int64(0)
+	// Parse the last message to get the version
-	for _, event := range events {
+	var event aether.Event
-		if event.Version > latestVersion {
+	if err := json.Unmarshal(msgs[0].Data, &event); err != nil {
-			latestVersion = event.Version
+		return 0, fmt.Errorf("failed to unmarshal last event: %w", err)
-		}
 	}
 
-	// Always repopulate cache with the fresh data just fetched
+	msgs[0].Ack()
-	// This ensures cache is in sync with actual state, whether from local writes
+	return event.Version, nil
-	// or external writes detected by version comparison
-	jes.versions[actorID] = latestVersion
-
-	return latestVersion, nil
 }
 
 // GetLatestSnapshot gets the most recent snapshot for an actor

store/jetstream_benchmark_test.go

@@ -0,0 +1,146 @@
+//go:build integration
+
+package store
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"git.flowmade.one/flowmade-one/aether"
+)
+
+// BenchmarkGetLatestVersion_WithManyEvents benchmarks GetLatestVersion performance
+// with a large number of events per actor.
+// This demonstrates the O(1) performance by showing that time doesn't increase
+// significantly with more events.
+func BenchmarkGetLatestVersion_WithManyEvents(b *testing.B) {
+	nc := getTestNATSConnection(&testing.T{})
+	if nc == nil {
+		b.Skip("NATS not available")
+		return
+	}
+	defer nc.Close()
+
+	store, err := NewJetStreamEventStore(nc, fmt.Sprintf("bench-getversion-%d", time.Now().UnixNano()))
+	if err != nil {
+		b.Fatalf("failed to create store: %v", err)
+	}
+
+	actorID := "actor-bench-test"
+
+	// Populate with 1000 events
+	for i := 1; i <= 1000; i++ {
+		event := &aether.Event{
+			ID:        fmt.Sprintf("evt-%d", i),
+			EventType: "BenchEvent",
+			ActorID:   actorID,
+			Version:   int64(i),
+			Data:      map[string]interface{}{"index": i},
+			Timestamp: time.Now(),
+		}
+		err := store.SaveEvent(event)
+		if err != nil {
+			b.Fatalf("SaveEvent failed for event %d: %v", i, err)
+		}
+	}
+
+	// Benchmark GetLatestVersion
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := store.GetLatestVersion(actorID)
+		if err != nil {
+			b.Fatalf("GetLatestVersion failed: %v", err)
+		}
+	}
+	b.StopTimer()
+}
+
+// BenchmarkGetLatestVersion_NoCache benchmarks GetLatestVersion without cache
+// to show that even uncached lookups are very fast due to DeliverLast optimization
+func BenchmarkGetLatestVersion_NoCache(b *testing.B) {
+	nc := getTestNATSConnection(&testing.T{})
+	if nc == nil {
+		b.Skip("NATS not available")
+		return
+	}
+	defer nc.Close()
+
+	store, err := NewJetStreamEventStore(nc, fmt.Sprintf("bench-nocache-%d", time.Now().UnixNano()))
+	if err != nil {
+		b.Fatalf("failed to create store: %v", err)
+	}
+
+	// Create a new store instance each iteration to bypass cache
+	actorID := "actor-bench-nocache"
+
+	// Populate with 1000 events
+	for i := 1; i <= 1000; i++ {
+		event := &aether.Event{
+			ID:        fmt.Sprintf("evt-%d", i),
+			EventType: "BenchEvent",
+			ActorID:   actorID,
+			Version:   int64(i),
+			Data:      map[string]interface{}{"index": i},
+			Timestamp: time.Now(),
+		}
+		err := store.SaveEvent(event)
+		if err != nil {
+			b.Fatalf("SaveEvent failed for event %d: %v", i, err)
+		}
+	}
+
+	// Benchmark GetLatestVersion without using cache (fresh instance)
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		// Create a new store to bypass version cache
+		newStore, err := NewJetStreamEventStore(nc, store.GetStreamName())
+		if err != nil {
+			b.Fatalf("failed to create new store: %v", err)
+		}
+		_, err = newStore.GetLatestVersion(actorID)
+		if err != nil {
+			b.Fatalf("GetLatestVersion failed: %v", err)
+		}
+	}
+	b.StopTimer()
+}
+
+// BenchmarkGetLatestVersion_SingleEvent benchmarks with minimal data
+func BenchmarkGetLatestVersion_SingleEvent(b *testing.B) {
+	nc := getTestNATSConnection(&testing.T{})
+	if nc == nil {
+		b.Skip("NATS not available")
+		return
+	}
+	defer nc.Close()
+
+	store, err := NewJetStreamEventStore(nc, fmt.Sprintf("bench-single-%d", time.Now().UnixNano()))
+	if err != nil {
+		b.Fatalf("failed to create store: %v", err)
+	}
+
+	actorID := "actor-single"
+
+	event := &aether.Event{
+		ID:        "evt-1",
+		EventType: "TestEvent",
+		ActorID:   actorID,
+		Version:   1,
+		Data:      map[string]interface{}{},
+		Timestamp: time.Now(),
+	}
+	err = store.SaveEvent(event)
+	if err != nil {
+		b.Fatalf("SaveEvent failed: %v", err)
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := store.GetLatestVersion(actorID)
+		if err != nil {
+			b.Fatalf("GetLatestVersion failed: %v", err)
+		}
+	}
+	b.StopTimer()
+}