tests: add testing around set and es aggregate

pkg/es/driver/mem-store/mem-store.go

@@ -89,9 +89,21 @@ func (m *eventLog) Append(ctx context.Context, events event.Events, version uint
 		for i := range events {
 			span.AddEvent(fmt.Sprintf("read event %d of %d", i, len(events)))
 
+			// --- clone event
+			e := events[i]
+			b, err := event.MarshalBinary(e)
+			if err != nil {
+				return err
+			}
+			e, err = event.UnmarshalBinary(ctx, b, e.EventMeta().Position)
+			if err != nil {
+				return err
+			}
+			// ---
+
 			pos := last + uint64(i) + 1
-			event.SetPosition(events[i], pos)
-			*stream = append(*stream, events[i])
+			event.SetPosition(e, pos)
+			*stream = append(*stream, e)
 		}
 
 		return nil
@@ -126,8 +138,17 @@ func (m *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Even
 		events = make([]event.Event, math.Abs(count))
 		for i := range events {
 			span.AddEvent(fmt.Sprintf("read event %d of %d", i, math.Abs(count)))
-			// ---
-			events[i] = (*stream)[start-1]
+
+			// --- clone event
+			e := (*stream)[start-1]
+			b, err := event.MarshalBinary(e)
+			if err != nil {
+				return err
+			}
+			events[i], err = event.UnmarshalBinary(ctx, b, e.EventMeta().Position)
+			if err != nil {
+				return err
+			}
 			// ---
 
 			if count > 0 {
@@ -140,6 +161,7 @@ func (m *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Even
 				break
 			}
 		}
+		event.SetStreamID(m.streamID, events...)
 
 		return nil
 	})
@@ -147,8 +169,6 @@ func (m *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Even
 		return nil, err
 	}
 
-	event.SetStreamID(m.streamID, events...)
-
 	return events, nil
 }
 

pkg/es/driver/projecter/projecter.go

@@ -71,8 +71,9 @@ func (w *wrapper) Append(ctx context.Context, events event.Events, version uint6
 			for i := range events {
 				e := events[i]
 				eventType := event.TypeOf(e)
-				streamID := e.EventMeta().StreamID
-				streamPos := e.EventMeta().Position
+				m := e.EventMeta()
+				streamID := m.StreamID
+				streamPos := m.Position
 
 				e1 := event.NewPtr(streamID, streamPos)
 				event.SetStreamID("$all", e1)