aether/store/memory.go

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.
// Panics if JSON marshaling/unmarshaling fails, which should never occur
// for a valid Event structure.
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, err := json.Marshal(event)
	if err != nil {
		panic("failed to marshal event: " + err.Error())
	}
	var copy aether.Event
	err = json.Unmarshal(data, &copy)
	if err != nil {
		panic("failed to unmarshal event: " + err.Error())
	}

	// 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 &copy
}

// 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"
}