feat: add subscriptions and rework interfaces

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

@@ -12,6 +12,7 @@ import (
 	"github.com/sour-is/ev/pkg/es"
 	"github.com/sour-is/ev/pkg/es/driver"
 	"github.com/sour-is/ev/pkg/es/event"
+	"github.com/sour-is/ev/pkg/locker"
 	"github.com/sour-is/ev/pkg/math"
 )
 
@@ -19,14 +20,17 @@ type diskStore struct {
 	path string
 }
 
-var _ driver.Driver = (*diskStore)(nil)
+const AppendOnly = es.AppendOnly
+const AllEvents = es.AllEvents
 
 func Init(ctx context.Context) error {
 	es.Register(ctx, "file", &diskStore{})
 	return nil
 }
 
-func (diskStore) Open(dsn string) (driver.EventStore, error) {
+var _ driver.Driver = (*diskStore)(nil)
+
+func (diskStore) Open(_ context.Context, dsn string) (driver.Driver, error) {
 	scheme, path, ok := strings.Cut(dsn, ":")
 	if !ok {
 		return nil, fmt.Errorf("expected scheme")
@@ -45,181 +49,151 @@ func (diskStore) Open(dsn string) (driver.EventStore, error) {
 
 	return &diskStore{path: path}, nil
 }
-
-func (es *diskStore) Save(ctx context.Context, agg event.Aggregate) (uint64, error) {
-	l, err := es.readLog(agg.StreamID())
-	if err != nil {
-		return 0, err
-	}
-
-	var last uint64
-	if last, err = l.LastIndex(); err != nil {
-		return 0, err
-	}
-	if agg.StreamVersion() != last {
-		return 0, fmt.Errorf("current version wrong %d != %d", agg.StreamVersion(), last)
-	}
-
-	events := agg.Events(true)
-
-	var b []byte
-	batch := &wal.Batch{}
-	for _, e := range events {
-		b, err = event.MarshalText(e)
-		if err != nil {
-			return 0, err
-		}
-
-		batch.Write(e.EventMeta().Position, b)
-	}
-
-	err = l.WriteBatch(batch)
-	if err != nil {
-		return 0, err
-	}
-	agg.Commit()
-
-	return uint64(len(events)), nil
+func (ds *diskStore) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) {
+	el := &eventLog{streamID: streamID}
+	l, err := wal.Open(filepath.Join(ds.path, streamID), wal.DefaultOptions)
+	el.events = locker.New(l)
+	return el, err
 }
-func (es *diskStore) Append(ctx context.Context, streamID string, events event.Events) (uint64, error) {
-	event.SetStreamID(streamID, events...)
 
-	l, err := es.readLog(streamID)
-	if err != nil {
-		return 0, err
-	}
-
-	var last uint64
-	if last, err = l.LastIndex(); err != nil {
-		return 0, err
-	}
-
-	var b []byte
-
-	batch := &wal.Batch{}
-	for i, e := range events {
-		b, err = event.MarshalText(e)
-		if err != nil {
-			return 0, err
-		}
-		pos := last + uint64(i) + 1
-		event.SetPosition(e, pos)
-
-		batch.Write(pos, b)
-	}
-
-	err = l.WriteBatch(batch)
-	if err != nil {
-		return 0, err
-	}
-	return uint64(len(events)), nil
+type eventLog struct {
+	streamID string
+	events   *locker.Locked[wal.Log]
 }
-func (es *diskStore) Load(ctx context.Context, agg event.Aggregate) error {
-	l, err := es.readLog(agg.StreamID())
-	if err != nil {
-		return err
-	}
 
-	var i, first, last uint64
+var _ driver.EventLog = (*eventLog)(nil)
 
-	if first, err = l.FirstIndex(); err != nil {
-		return err
-	}
-	if last, err = l.LastIndex(); err != nil {
-		return err
-	}
-	if first == 0 || last == 0 {
-		return nil
-	}
+func (es *eventLog) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) {
+	event.SetStreamID(es.streamID, events...)
 
-	var b []byte
-	events := make([]event.Event, last-i)
-	for i = 0; first+i <= last; i++ {
-		b, err = l.Read(first + i)
+	var count uint64
+	err := es.events.Modify(ctx, func(l *wal.Log) error {
+		last, err := l.LastIndex()
 		if err != nil {
 			return err
 		}
-		events[i], err = event.UnmarshalText(ctx, b, first+i)
-		if err != nil {
-			return err
+
+		if version != AppendOnly && version != last {
+			return fmt.Errorf("current version wrong %d != %d", version, last)
 		}
-	}
-	event.Append(agg, events...)
 
-	return nil
-}
-func (es *diskStore) Read(ctx context.Context, streamID string, pos, count int64) (event.Events, error) {
-	l, err := es.readLog(streamID)
-	if err != nil {
-		return nil, err
-	}
-
-	var first, last, start uint64
-	if first, err = l.FirstIndex(); err != nil {
-		return nil, err
-	}
-	if last, err = l.LastIndex(); err != nil {
-		return nil, err
-	}
-	if first == 0 || last == 0 {
-		return nil, nil
-	}
-
-	switch {
-	case pos >= 0:
-		start = first + uint64(pos)
-		if pos == 0 && count < 0 {
-			count = -count // if pos=0 assume forward count.
-		}
-	case pos < 0:
-		start = uint64(int64(last) + pos + 1)
-		if pos == -1 && count > 0 {
-			count = -count // if pos=-1 assume backward count.
-		}
-	}
-
-	events := make([]event.Event, math.Abs(count))
-	for i := range events {
 		var b []byte
 
-		b, err = l.Read(start)
+		batch := &wal.Batch{}
+		for i, e := range events {
+			b, err = event.MarshalText(e)
+			if err != nil {
+				return err
+			}
+			pos := last + uint64(i) + 1
+			event.SetPosition(e, pos)
+
+			batch.Write(pos, b)
+		}
+
+		count = uint64(len(events))
+		return l.WriteBatch(batch)
+	})
+
+	return count, err
+}
+func (es *eventLog) Read(ctx context.Context, pos, count int64) (event.Events, error) {
+	var events event.Events
+
+	err := es.events.Modify(ctx, func(stream *wal.Log) error {
+		first, err := stream.FirstIndex()
 		if err != nil {
-			return events, err
+			return err
 		}
-		events[i], err = event.UnmarshalText(ctx, b, start)
+		last, err := stream.LastIndex()
 		if err != nil {
-			return events, err
+			return err
+		}
+		// ---
+		if first == 0 || last == 0 {
+			return nil
 		}
 
-		if count > 0 {
-			start += 1
-		} else {
-			start -= 1
+		if count == AllEvents {
+			count = int64(first - last)
 		}
-		if start < first || start > last {
-			events = events[:i+1]
-			break
-		}
-	}
-	event.SetStreamID(streamID, events...)
 
-	return events, nil
-}
-func (es *diskStore) FirstIndex(ctx context.Context, streamID string) (uint64, error) {
-	l, err := es.readLog(streamID)
+		var start uint64
+
+		switch {
+		case pos >= 0 && count > 0:
+			start = first + uint64(pos)
+		case pos < 0 && count > 0:
+			start = uint64(int64(last) + pos + 1)
+
+		case pos >= 0 && count < 0:
+			start = first + uint64(pos)
+			if pos > 1 {
+				start -= 2 // if pos is positive and count negative start before
+			}
+			if pos <= 1 {
+				return nil // if pos is one or zero and negative count nothing to return
+			}
+		case pos < 0 && count < 0:
+			start = uint64(int64(last) + pos)
+		}
+		if start >= last {
+			return nil // if start is after last and positive count nothing to return
+		}
+
+		events = make([]event.Event, math.Abs(count))
+		for i := range events {
+			// ---
+			var b []byte
+			b, err = stream.Read(start)
+			if err != nil {
+				return err
+			}
+			events[i], err = event.UnmarshalText(ctx, b, start)
+			if err != nil {
+				return err
+			}
+			// ---
+
+			if count > 0 {
+				start += 1
+			} else {
+				start -= 1
+			}
+			if start < first || start > last {
+				events = events[:i+1]
+				break
+			}
+		}
+		return nil
+	})
 	if err != nil {
-		return 0, err
+		return nil, err
 	}
-	return l.FirstIndex()
-}
-func (es *diskStore) LastIndex(ctx context.Context, streamID string) (uint64, error) {
-	l, err := es.readLog(streamID)
-	if err != nil {
-		return 0, err
-	}
-	return l.LastIndex()
-}
 
-func (es *diskStore) readLog(name string) (*wal.Log, error) {
-	return wal.Open(filepath.Join(es.path, name), wal.DefaultOptions)
+	event.SetStreamID(es.streamID, events...)
+
+	return events, err
+}
+func (es *eventLog) FirstIndex(ctx context.Context) (uint64, error) {
+	var idx uint64
+	var err error
+
+	err = es.events.Modify(ctx, func(events *wal.Log) error {
+		idx, err = events.FirstIndex()
+		return err
+	})
+
+	return idx, err
+}
+func (es *eventLog) LastIndex(ctx context.Context) (uint64, error) {
+	var idx uint64
+	var err error
+
+	err = es.events.Modify(ctx, func(events *wal.Log) error {
+		idx, err = events.LastIndex()
+		return err
+	})
+
+	return idx, err
 }

pkg/es/driver/driver.go

@@ -7,14 +7,24 @@ import (
 )
 
 type Driver interface {
-	Open(string) (EventStore, error)
+	Open(ctx context.Context, dsn string) (Driver, error)
+	EventLog(ctx context.Context, streamID string) (EventLog, error)
 }
 
-type EventStore interface {
-	Save(ctx context.Context, agg event.Aggregate) (uint64, error)
-	Load(ctx context.Context, agg event.Aggregate) error
-	Read(ctx context.Context, streamID string, pos, count int64) (event.Events, error)
-	Append(ctx context.Context, streamID string, events event.Events) (uint64, error)
-	FirstIndex(ctx context.Context, streamID string) (uint64, error)
-	LastIndex(ctx context.Context, streamID string) (uint64, error)
+type EventLog interface {
+	Read(ctx context.Context, pos, count int64) (event.Events, error)
+	Append(ctx context.Context, events event.Events, version uint64) (uint64, error)
+	FirstIndex(ctx context.Context) (uint64, error)
+	LastIndex(ctx context.Context) (uint64, error)
+}
+
+type Subscription interface {
+	Recv(context.Context) bool
+	Events(context.Context) (event.Events, error)
+	Close(context.Context) error
+}
+
+type EventStream interface {
+	Subscribe(ctx context.Context, streamID string) (Subscription, error)
+	Send(ctx context.Context, streamID string, events event.Events) error
 }

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

@@ -12,84 +12,113 @@ import (
 )
 
 type state struct {
-	streams map[string]event.Events
+	streams map[string]*locker.Locked[event.Events]
+}
+type eventLog struct {
+	streamID string
+	events   *locker.Locked[event.Events]
 }
-
 type memstore struct {
 	state *locker.Locked[state]
 }
 
-var _ driver.Driver = (*memstore)(nil)
+const AppendOnly = es.AppendOnly
+const AllEvents = es.AllEvents
 
 func Init(ctx context.Context) {
 	es.Register(ctx, "mem", &memstore{})
 }
 
-func (memstore) Open(name string) (driver.EventStore, error) {
-	s := &state{streams: make(map[string]event.Events)}
+var _ driver.Driver = (*memstore)(nil)
+
+func (memstore) Open(_ context.Context, name string) (driver.Driver, error) {
+	s := &state{streams: make(map[string]*locker.Locked[event.Events])}
 	return &memstore{locker.New(s)}, nil
 }
+func (m *memstore) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) {
+	el := &eventLog{streamID: streamID}
+
+	err := m.state.Modify(ctx, func(state *state) error {
+		l, ok := state.streams[streamID]
+		if !ok {
+			l = locker.New(&event.Events{})
+			state.streams[streamID] = l
+		}
+		el.events = l
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return el, err
+}
+
+var _ driver.EventLog = (*eventLog)(nil)
 
 // Append implements driver.EventStore
-func (m *memstore) Append(ctx context.Context, streamID string, events event.Events) (uint64, error) {
-	event.SetStreamID(streamID, events...)
+func (m *eventLog) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) {
+	event.SetStreamID(m.streamID, events...)
+
+	return uint64(len(events)), m.events.Modify(ctx, func(stream *event.Events) error {
+		last := uint64(len(*stream))
+		if version != AppendOnly && version != last {
+			return fmt.Errorf("current version wrong %d != %d", version, last)
+		}
 
-	return uint64(len(events)), m.state.Modify(ctx, func(state *state) error {
-		stream := state.streams[streamID]
-		last := uint64(len(stream))
 		for i := range events {
 			pos := last + uint64(i) + 1
 			event.SetPosition(events[i], pos)
-			stream = append(stream, events[i])
-			state.streams[streamID] = stream
+			*stream = append(*stream, events[i])
 		}
 
 		return nil
 	})
 }
 
-// Load implements driver.EventStore
-func (m *memstore) Load(ctx context.Context, agg event.Aggregate) error {
-	return m.state.Modify(ctx, func(state *state) error {
-		events := state.streams[agg.StreamID()]
-		event.SetStreamID(agg.StreamID(), events...)
-		agg.ApplyEvent(events...)
-		return nil
-	})
-}
-
 // Read implements driver.EventStore
-func (m *memstore) Read(ctx context.Context, streamID string, pos int64, count int64) (event.Events, error) {
-	events := make([]event.Event, math.Abs(count))
-
-	err := m.state.Modify(ctx, func(state *state) error {
-		stream := state.streams[streamID]
-
-		var first, last, start uint64
-		first = stream.First().EventMeta().Position
-		last = stream.Last().EventMeta().Position
+func (es *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Events, error) {
+	var events event.Events
 
+	err := es.events.Modify(ctx, func(stream *event.Events) error {
+		first := stream.First().EventMeta().Position
+		last := stream.Last().EventMeta().Position
+		// ---
 		if first == 0 || last == 0 {
-			events = events[:0]
-
 			return nil
 		}
 
-		switch {
-		case pos >= 0:
-			start = first + uint64(pos)
-			if pos == 0 && count < 0 {
-				count = -count // if pos=0 assume forward count.
-			}
-		case pos < 0:
-			start = uint64(int64(last) + pos + 1)
-			if pos == -1 && count > 0 {
-				count = -count // if pos=-1 assume backward count.
-			}
+		if count == AllEvents {
+			count = int64(first - last)
 		}
 
+		var start uint64
+
+		switch {
+		case pos >= 0 && count > 0:
+			start = first + uint64(pos)
+		case pos < 0 && count > 0:
+			start = uint64(int64(last) + pos + 1)
+
+		case pos >= 0 && count < 0:
+			start = first + uint64(pos)
+			if pos > 1 {
+				start -= 2 // if pos is positive and count negative start before
+			}
+			if pos <= 1 {
+				return nil // if pos is one or zero and negative count nothing to return
+			}
+		case pos < 0 && count < 0:
+			start = uint64(int64(last) + pos)
+		}
+		if start >= last {
+			return nil // if start is after last and positive count nothing to return
+		}
+
+		events = make([]event.Event, math.Abs(count))
 		for i := range events {
-			events[i] = stream[start-1]
+			// ---
+			events[i] = (*stream)[start-1]
+			// ---
 
 			if count > 0 {
 				start += 1
@@ -108,51 +137,19 @@ func (m *memstore) Read(ctx context.Context, streamID string, pos int64, count i
 		return nil, err
 	}
 
+	event.SetStreamID(es.streamID, events...)
+
 	return events, nil
 }
 
-// Save implements driver.EventStore
-func (m *memstore) Save(ctx context.Context, agg event.Aggregate) (uint64, error) {
-	events := agg.Events(true)
-	event.SetStreamID(agg.StreamID(), events...)
-
-	err := m.state.Modify(ctx, func(state *state) error {
-		stream := state.streams[agg.StreamID()]
-
-		last := uint64(len(stream))
-		if agg.StreamVersion() != last {
-			return fmt.Errorf("current version wrong %d != %d", agg.StreamVersion(), last)
-		}
-
-		for i := range events {
-			pos := last + uint64(i) + 1
-			event.SetPosition(events[i], pos)
-			stream = append(stream, events[i])
-		}
-
-		state.streams[agg.StreamID()] = stream
-		return nil
-	})
-	if err != nil {
-		return 0, err
-	}
-	agg.Commit()
-
-	return uint64(len(events)), nil
+// FirstIndex for the streamID
+func (m *eventLog) FirstIndex(ctx context.Context) (uint64, error) {
+	events, err := m.events.Copy(ctx)
+	return events.First().EventMeta().Position, err
 }
 
-func (m *memstore) FirstIndex(ctx context.Context, streamID string) (uint64, error) {
-	stream, err := m.state.Copy(ctx)
-	if err != nil {
-		return 0, err
-	}
-	return stream.streams[streamID].First().EventMeta().Position, nil
-}
-func (m *memstore) LastIndex(ctx context.Context, streamID string) (uint64, error) {
-	stream, err := m.state.Copy(ctx)
-	if err != nil {
-		return 0, err
-	}
-	return stream.streams[streamID].Last().EventMeta().Position, nil
-
+// LastIndex for the streamID
+func (m *eventLog) LastIndex(ctx context.Context) (uint64, error) {
+	events, err := m.events.Copy(ctx)
+	return events.Last().EventMeta().Position, err
 }

pkg/es/driver/streamer/streamer.go

@@ -0,0 +1,203 @@
+package streamer
+
+import (
+	"context"
+	"log"
+
+	"github.com/sour-is/ev/pkg/es"
+	"github.com/sour-is/ev/pkg/es/driver"
+	"github.com/sour-is/ev/pkg/es/event"
+	"github.com/sour-is/ev/pkg/locker"
+)
+
+type state struct {
+	subscribers map[string][]*subscription
+}
+
+type streamer struct {
+	state *locker.Locked[state]
+	up    driver.Driver
+}
+
+func New(ctx context.Context) *streamer {
+	return &streamer{state: locker.New(&state{subscribers: map[string][]*subscription{}})}
+}
+
+var _ es.Option = (*streamer)(nil)
+
+func (s *streamer) Apply(e *es.EventStore) {
+	s.up = e.Driver
+	e.Driver = s
+}
+func (s *streamer) Unwrap() driver.Driver {
+	return s.up
+}
+
+var _ driver.Driver = (*streamer)(nil)
+
+func (s *streamer) Open(ctx context.Context, dsn string) (driver.Driver, error) {
+	return s.up.Open(ctx, dsn)
+}
+func (s *streamer) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) {
+	l, err := s.up.EventLog(ctx, streamID)
+	return &wrapper{streamID, l, s}, err
+}
+
+var _ driver.EventStream = (*streamer)(nil)
+
+func (s *streamer) Subscribe(ctx context.Context, streamID string) (driver.Subscription, error) {
+	log.Println("subscribe", streamID)
+	events, err := s.up.EventLog(ctx, streamID)
+	if err != nil {
+		return nil, err
+	}
+	sub := &subscription{topic: streamID, events: events}
+	sub.position = locker.New(&position{
+		size: es.AllEvents,
+	})
+	sub.unsub = s.delete(streamID, sub)
+
+	return sub, s.state.Modify(ctx, func(state *state) error {
+		state.subscribers[streamID] = append(state.subscribers[streamID], sub)
+		log.Println("subs=", len(state.subscribers[streamID]))
+		return nil
+	})
+}
+func (s *streamer) Send(ctx context.Context, streamID string, events event.Events) error {
+	log.Println("send", streamID, len(events))
+	return s.state.Modify(ctx, func(state *state) error {
+		for _, sub := range state.subscribers[streamID] {
+			return sub.position.Modify(ctx, func(position *position) error {
+				position.size = int64(events.Last().EventMeta().Position - uint64(position.idx))
+
+				if position.wait != nil {
+					close(position.wait)
+					position.wait = nil
+				}
+				return nil
+			})
+		}
+		return nil
+	})
+}
+
+func (s *streamer) delete(streamID string, sub *subscription) func(context.Context) error {
+	return func(ctx context.Context) error {
+		log.Println("unsub", streamID)
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		return s.state.Modify(ctx, func(state *state) error {
+			lis := state.subscribers[streamID]
+			for i := range lis {
+				if lis[i] == sub {
+					lis[i] = lis[len(lis)-1]
+					state.subscribers[streamID] = lis[:len(lis)-1]
+					log.Println("subs=", len(state.subscribers[streamID]))
+
+					return nil
+				}
+			}
+			return nil
+		})
+	}
+}
+
+type wrapper struct {
+	topic    string
+	up       driver.EventLog
+	streamer *streamer
+}
+
+var _ driver.EventLog = (*wrapper)(nil)
+
+func (w *wrapper) Read(ctx context.Context, pos int64, count int64) (event.Events, error) {
+	return w.up.Read(ctx, pos, count)
+}
+
+func (w *wrapper) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) {
+	i, err := w.up.Append(ctx, events, version)
+	if err != nil {
+		return i, err
+	}
+	return i, w.streamer.Send(ctx, w.topic, events)
+}
+
+func (w *wrapper) FirstIndex(ctx context.Context) (uint64, error) {
+	return w.up.FirstIndex(ctx)
+}
+
+func (w *wrapper) LastIndex(ctx context.Context) (uint64, error) {
+	return w.up.LastIndex(ctx)
+}
+
+type position struct {
+	size int64
+	idx  int64
+	wait chan struct{}
+}
+
+type subscription struct {
+	topic string
+
+	position *locker.Locked[position]
+
+	events driver.EventLog
+	unsub  func(context.Context) error
+}
+
+func (s *subscription) Recv(ctx context.Context) bool {
+	var wait func(context.Context) bool
+	log.Println("recv more")
+	err := s.position.Modify(ctx, func(position *position) error {
+		if position.size == es.AllEvents {
+			return nil
+		}
+		if position.size == 0 {
+			position.wait = make(chan struct{})
+			wait = func(ctx context.Context) bool {
+				log.Println("waiting", s.topic)
+				select {
+				case <-position.wait:
+					log.Println("got some")
+
+					return true
+				case <-ctx.Done():
+					log.Println("got cancel")
+
+					return false
+				}
+			}
+		}
+		position.idx += position.size
+		position.size = 0
+		return nil
+	})
+	if err != nil {
+		return false
+	}
+
+	if wait != nil {
+		return wait(ctx)
+	}
+
+	return true
+}
+func (s *subscription) Events(ctx context.Context) (event.Events, error) {
+	var events event.Events
+	log.Println("get events")
+	return events, s.position.Modify(ctx, func(position *position) error {
+		var err error
+		events, err = s.events.Read(ctx, int64(position.idx), position.size)
+		log.Printf("got events=%d %#v", len(events), position)
+		position.size = int64(len(events))
+		if len(events) > 0 {
+			position.idx = int64(events.First().EventMeta().Position - 1)
+			log.Println(position, events.First())
+		}
+		return err
+	})
+}
+func (s *subscription) Close(ctx context.Context) error {
+	return s.unsub(ctx)
+}