Add event versioning validation
All checks were successful
CI / build (pull_request) Successful in 16s
CI / build (push) Successful in 15s

- Add ErrVersionConflict error type and VersionConflictError for detailed
  conflict information
- Implement version validation in InMemoryEventStore.SaveEvent that rejects
  events with version <= current latest version
- Implement version validation in JetStreamEventStore.SaveEvent with version
  caching for performance
- Add comprehensive tests for version conflict detection including concurrent
  writes to same actor
- Document versioning semantics in EventStore interface and CLAUDE.md

This ensures events have monotonically increasing versions per actor and
provides clear error messages for version conflicts, enabling optimistic
concurrency control patterns.

Closes #6

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit was merged in pull request #34.
This commit is contained in:
2026-01-09 17:56:50 +01:00
parent a269da4520
commit 02847bdaf5
5 changed files with 401 additions and 19 deletions

View File

@@ -4,6 +4,7 @@ import (
"encoding/json"
"fmt"
"strings"
"sync"
"time"
"git.flowmade.one/flowmade-one/aether"
@@ -14,6 +15,8 @@ import (
type JetStreamEventStore struct {
js nats.JetStreamContext
streamName string
mu sync.Mutex // Protects version checks during SaveEvent
versions map[string]int64 // actorID -> latest version cache
}
// NewJetStreamEventStore creates a new JetStream-based event store
@@ -41,11 +44,32 @@ func NewJetStreamEventStore(natsConn *nats.Conn, streamName string) (*JetStreamE
return &JetStreamEventStore{
js: js,
streamName: streamName,
versions: make(map[string]int64),
}, nil
}
// SaveEvent persists an event to JetStream
// SaveEvent persists an event to JetStream.
// Returns VersionConflictError if the event's version is not strictly greater
// than the current latest version for the actor.
func (jes *JetStreamEventStore) SaveEvent(event *aether.Event) error {
jes.mu.Lock()
defer jes.mu.Unlock()
// Get current latest version for this actor
currentVersion, err := jes.getLatestVersionLocked(event.ActorID)
if err != nil {
return fmt.Errorf("failed to get latest version: %w", err)
}
// Validate version is strictly greater than current
if event.Version <= currentVersion {
return &aether.VersionConflictError{
ActorID: event.ActorID,
AttemptedVersion: event.Version,
CurrentVersion: currentVersion,
}
}
// Serialize event to JSON
data, err := json.Marshal(event)
if err != nil {
@@ -64,11 +88,50 @@ func (jes *JetStreamEventStore) SaveEvent(event *aether.Event) error {
return fmt.Errorf("failed to publish event to JetStream: %w", err)
}
// Update version cache
jes.versions[event.ActorID] = event.Version
return nil
}
// getLatestVersionLocked returns the latest version for an actor.
// Caller must hold jes.mu.
func (jes *JetStreamEventStore) getLatestVersionLocked(actorID string) (int64, error) {
// Check cache first
if version, ok := jes.versions[actorID]; ok {
return version, nil
}
// Fetch from JetStream
events, err := jes.getEventsInternal(actorID, 0)
if err != nil {
return 0, err
}
if len(events) == 0 {
return 0, nil
}
latestVersion := int64(0)
for _, event := range events {
if event.Version > latestVersion {
latestVersion = event.Version
}
}
// Update cache
jes.versions[actorID] = latestVersion
return latestVersion, nil
}
// GetEvents retrieves all events for an actor since a version
func (jes *JetStreamEventStore) GetEvents(actorID string, fromVersion int64) ([]*aether.Event, error) {
return jes.getEventsInternal(actorID, fromVersion)
}
// getEventsInternal is the internal implementation of GetEvents
func (jes *JetStreamEventStore) getEventsInternal(actorID string, fromVersion int64) ([]*aether.Event, error) {
// Create subject filter for this actor
subject := fmt.Sprintf("%s.events.%s.%s",
jes.streamName,