112 lines
2.8 KiB
Go
112 lines
2.8 KiB
Go
package store
|
|
|
|
import (
|
|
"fmt"
|
|
"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),
|
|
}
|
|
}
|
|
|
|
// SaveEvent saves an event to the in-memory store
|
|
func (es *InMemoryEventStore) SaveEvent(event *aether.Event) error {
|
|
es.mu.Lock()
|
|
defer es.mu.Unlock()
|
|
|
|
if _, exists := es.events[event.ActorID]; !exists {
|
|
es.events[event.ActorID] = make([]*aether.Event, 0)
|
|
}
|
|
es.events[event.ActorID] = append(es.events[event.ActorID], event)
|
|
return nil
|
|
}
|
|
|
|
// GetEvents retrieves events for an actor from a specific version
|
|
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 {
|
|
filteredEvents = append(filteredEvents, 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 fmt.Errorf("snapshot cannot be nil")
|
|
}
|
|
|
|
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
|
|
}
|