package store import ( "encoding/json" "sync" "git.flowmade.one/flowmade-one/aether" ) // InMemoryEventStore provides a simple in-memory event store for testing type InMemoryEventStore struct { mu sync.RWMutex events map[string][]*aether.Event // actorID -> events snapshots map[string][]*aether.ActorSnapshot // actorID -> snapshots (sorted by version) } // NewInMemoryEventStore creates a new in-memory event store func NewInMemoryEventStore() *InMemoryEventStore { return &InMemoryEventStore{ events: make(map[string][]*aether.Event), snapshots: make(map[string][]*aether.ActorSnapshot), } } // deepCopyEvent creates a deep copy of an event to ensure immutability. // This prevents modifications to the event after it's been persisted. func deepCopyEvent(event *aether.Event) *aether.Event { // Use JSON marshaling/unmarshaling for a complete deep copy // This ensures all nested structures (maps, slices) are copied data, _ := json.Marshal(event) var copy aether.Event json.Unmarshal(data, ©) // Preserve empty metadata maps (JSON unmarshal converts empty map to nil) if event.Metadata != nil && len(event.Metadata) == 0 && copy.Metadata == nil { copy.Metadata = make(map[string]string) } // Preserve empty data maps if event.Data != nil && len(event.Data) == 0 && copy.Data == nil { copy.Data = make(map[string]interface{}) } return © } // SaveEvent saves an event to the in-memory store. // Returns VersionConflictError if the event's version is not strictly greater // than the current latest version for the actor. // The event is deep-copied before storage to ensure immutability. func (es *InMemoryEventStore) SaveEvent(event *aether.Event) error { es.mu.Lock() defer es.mu.Unlock() // Get current latest version for this actor currentVersion := int64(0) if events, exists := es.events[event.ActorID]; exists { for _, e := range events { if e.Version > currentVersion { currentVersion = e.Version } } } // Validate version is strictly greater than current if event.Version <= currentVersion { return &aether.VersionConflictError{ ActorID: event.ActorID, AttemptedVersion: event.Version, CurrentVersion: currentVersion, } } if _, exists := es.events[event.ActorID]; !exists { es.events[event.ActorID] = make([]*aether.Event, 0) } // Deep copy the event before storing to ensure immutability storedEvent := deepCopyEvent(event) es.events[event.ActorID] = append(es.events[event.ActorID], storedEvent) return nil } // GetEvents retrieves events for an actor from a specific version // Returns deep copies of events to ensure immutability func (es *InMemoryEventStore) GetEvents(actorID string, fromVersion int64) ([]*aether.Event, error) { es.mu.RLock() defer es.mu.RUnlock() events, exists := es.events[actorID] if !exists { return []*aether.Event{}, nil } var filteredEvents []*aether.Event for _, event := range events { if event.Version >= fromVersion { // Return a deep copy to ensure immutability filteredEvents = append(filteredEvents, deepCopyEvent(event)) } } return filteredEvents, nil } // GetLatestVersion returns the latest version for an actor func (es *InMemoryEventStore) GetLatestVersion(actorID string) (int64, error) { es.mu.RLock() defer es.mu.RUnlock() events, exists := es.events[actorID] if !exists || len(events) == 0 { return 0, nil } latestVersion := int64(0) for _, event := range events { if event.Version > latestVersion { latestVersion = event.Version } } return latestVersion, nil } // SaveSnapshot saves a snapshot to the in-memory store func (es *InMemoryEventStore) SaveSnapshot(snapshot *aether.ActorSnapshot) error { if snapshot == nil { return &snapshotNilError{} } es.mu.Lock() defer es.mu.Unlock() if _, exists := es.snapshots[snapshot.ActorID]; !exists { es.snapshots[snapshot.ActorID] = make([]*aether.ActorSnapshot, 0) } es.snapshots[snapshot.ActorID] = append(es.snapshots[snapshot.ActorID], snapshot) return nil } // GetLatestSnapshot returns the most recent snapshot for an actor func (es *InMemoryEventStore) GetLatestSnapshot(actorID string) (*aether.ActorSnapshot, error) { es.mu.RLock() defer es.mu.RUnlock() snapshots, exists := es.snapshots[actorID] if !exists || len(snapshots) == 0 { return nil, nil } var latest *aether.ActorSnapshot for _, snapshot := range snapshots { if latest == nil || snapshot.Version > latest.Version { latest = snapshot } } return latest, nil } // snapshotNilError is returned when attempting to save a nil snapshot type snapshotNilError struct{} func (e *snapshotNilError) Error() string { return "snapshot cannot be nil" }