fix: improve paging and subscriptions

api/gql_ev/msgbus.graphqls

@@ -2,7 +2,8 @@ extend type Query {
     posts(streamID: String! paging: PageInput): Connection!
 }
 extend type Subscription {
-    postAdded(streamID: String!): PostEvent
+    """after == 0 start from begining, after == -1 start from end"""
+    postAdded(streamID: String! after: Int! = -1): PostEvent
 }
 type PostEvent implements Edge {
     id: ID!

api/gql_ev/resolver.go

@@ -3,7 +3,6 @@ package gql_ev
 import (
 	"context"
 	"fmt"
-	"log"
 	"time"
 
 	"github.com/sour-is/ev/pkg/es"
@@ -66,13 +65,13 @@ func (r *Resolver) Posts(ctx context.Context, streamID string, paging *PageInput
 	}, nil
 }
 
-func (r *Resolver) PostAdded(ctx context.Context, streamID string) (<-chan *PostEvent, error) {
+func (r *Resolver) PostAdded(ctx context.Context, streamID string, after int64) (<-chan *PostEvent, error) {
 	es := r.es.EventStream()
 	if es == nil {
 		return nil, fmt.Errorf("EventStore does not implement streaming")
 	}
 
-	sub, err := es.Subscribe(ctx, streamID)
+	sub, err := es.Subscribe(ctx, streamID, after)
 	if err != nil {
 		return nil, err
 	}
@@ -83,7 +82,7 @@ func (r *Resolver) PostAdded(ctx context.Context, streamID string) (<-chan *Post
 		defer func() {
 			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 			defer cancel()
-			log.Print(sub.Close(ctx))
+			sub.Close(ctx)
 		}()
 
 		for sub.Recv(ctx) {

gqlgen.yml

@@ -54,12 +54,12 @@ models:
       - github.com/99designs/gqlgen/graphql.Uint32
   Int:
     model:
-      - github.com/99designs/gqlgen/graphql.Int
       - github.com/99designs/gqlgen/graphql.Int64
       - github.com/99designs/gqlgen/graphql.Int32
-      - github.com/99designs/gqlgen/graphql.Uint
+      - github.com/99designs/gqlgen/graphql.Int
       - github.com/99designs/gqlgen/graphql.Uint64
       - github.com/99designs/gqlgen/graphql.Uint32
+      - github.com/99designs/gqlgen/graphql.Uint
   Time:
     model:
       - github.com/99designs/gqlgen/graphql.Time

internal/graph/generated/generated.go

@@ -81,7 +81,7 @@ type ComplexityRoot struct {
 	}
 
 	Subscription struct {
-		PostAdded func(childComplexity int, streamID string) int
+		PostAdded func(childComplexity int, streamID string, after int64) int
 	}
 
 	_Service struct {
@@ -93,7 +93,7 @@ type QueryResolver interface {
 	Posts(ctx context.Context, streamID string, paging *gql_ev.PageInput) (*gql_ev.Connection, error)
 }
 type SubscriptionResolver interface {
-	PostAdded(ctx context.Context, streamID string) (<-chan *gql_ev.PostEvent, error)
+	PostAdded(ctx context.Context, streamID string, after int64) (<-chan *gql_ev.PostEvent, error)
 }
 
 type executableSchema struct {
@@ -245,7 +245,7 @@ func (e *executableSchema) Complexity(typeName, field string, childComplexity in
 			return 0, false
 		}
 
-		return e.complexity.Subscription.PostAdded(childComplexity, args["streamID"].(string)), true
+		return e.complexity.Subscription.PostAdded(childComplexity, args["streamID"].(string), args["after"].(int64)), true
 
 	case "_Service.sdl":
 		if e.complexity._Service.SDL == nil {
@@ -372,7 +372,8 @@ directive @goTag(
     posts(streamID: String! paging: PageInput): Connection!
 }
 extend type Subscription {
-    postAdded(streamID: String!): PostEvent
+    """after == 0 start from begining, after == -1 start from end"""
+    postAdded(streamID: String! after: Int! = -1): PostEvent
 }
 type PostEvent implements Edge {
     id: ID!
@@ -461,6 +462,15 @@ func (ec *executionContext) field_Subscription_postAdded_args(ctx context.Contex
 		}
 	}
 	args["streamID"] = arg0
+	var arg1 int64
+	if tmp, ok := rawArgs["after"]; ok {
+		ctx := graphql.WithPathContext(ctx, graphql.NewPathWithField("after"))
+		arg1, err = ec.unmarshalNInt2int64(ctx, tmp)
+		if err != nil {
+			return nil, err
+		}
+	}
+	args["after"] = arg1
 	return args, nil
 }
 
@@ -1434,7 +1444,7 @@ func (ec *executionContext) _Subscription_postAdded(ctx context.Context, field g
 	}()
 	resTmp, err := ec.ResolverMiddleware(ctx, func(rctx context.Context) (interface{}, error) {
 		ctx = rctx // use context from middleware stack in children
-		return ec.resolvers.Subscription().PostAdded(rctx, fc.Args["streamID"].(string))
+		return ec.resolvers.Subscription().PostAdded(rctx, fc.Args["streamID"].(string), fc.Args["after"].(int64))
 	})
 	if err != nil {
 		ec.Error(ctx, err)
@@ -4130,6 +4140,21 @@ func (ec *executionContext) marshalNID2string(ctx context.Context, sel ast.Selec
 	return res
 }
 
+func (ec *executionContext) unmarshalNInt2int64(ctx context.Context, v interface{}) (int64, error) {
+	res, err := graphql.UnmarshalInt64(v)
+	return res, graphql.ErrorOnPath(ctx, err)
+}
+
+func (ec *executionContext) marshalNInt2int64(ctx context.Context, sel ast.SelectionSet, v int64) graphql.Marshaler {
+	res := graphql.MarshalInt64(v)
+	if res == graphql.Null {
+		if !graphql.HasFieldError(ctx, graphql.GetFieldContext(ctx)) {
+			ec.Errorf(ctx, "the requested element is null which the schema does not allow")
+		}
+	}
+	return res
+}
+
 func (ec *executionContext) unmarshalNInt2uint64(ctx context.Context, v interface{}) (uint64, error) {
 	res, err := graphql.UnmarshalUint64(v)
 	return res, graphql.ErrorOnPath(ctx, err)

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

@@ -114,31 +114,9 @@ func (es *eventLog) Read(ctx context.Context, pos, count int64) (event.Events, e
 			return nil
 		}
 
-		if count == AllEvents {
+		start, count := math.PagerBox(first, last, pos, count)
-			count = int64(first - last)
+		if count == 0 {
-		}
+			return nil
-
-		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))

pkg/es/driver/driver.go

@@ -25,6 +25,6 @@ type Subscription interface {
 }
 
 type EventStream interface {
-	Subscribe(ctx context.Context, streamID string) (Subscription, error)
+	Subscribe(ctx context.Context, streamID string, start int64) (Subscription, error)
 	Send(ctx context.Context, streamID string, events event.Events) error
 }

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

@@ -87,31 +87,9 @@ func (es *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Eve
 			return nil
 		}
 
-		if count == AllEvents {
+		start, count := math.PagerBox(first, last, pos, count)
-			count = int64(first - last)
+		if count == 0 {
-		}
+			return nil
-
-		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))

pkg/es/driver/streamer/streamer.go

@@ -2,7 +2,6 @@ package streamer
 
 import (
 	"context"
-	"log"
 
 	"github.com/sour-is/ev/pkg/es"
 	"github.com/sour-is/ev/pkg/es/driver"
@@ -45,26 +44,24 @@ func (s *streamer) EventLog(ctx context.Context, streamID string) (driver.EventL
 
 var _ driver.EventStream = (*streamer)(nil)
 
-func (s *streamer) Subscribe(ctx context.Context, streamID string) (driver.Subscription, error) {
+func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64) (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{
+		idx:  start,
 		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 {
@@ -83,7 +80,6 @@ func (s *streamer) Send(ctx context.Context, streamID string, events event.Event
 
 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
 		}
@@ -93,7 +89,6 @@ func (s *streamer) delete(streamID string, sub *subscription) func(context.Conte
 				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
 				}
@@ -148,7 +143,6 @@ type subscription struct {
 
 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
@@ -156,14 +150,11 @@ func (s *subscription) Recv(ctx context.Context) bool {
 		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
 				}
@@ -185,15 +176,12 @@ func (s *subscription) Recv(ctx context.Context) bool {
 }
 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
 	})

pkg/locker/locker.go

@@ -2,7 +2,6 @@ package locker
 
 import (
 	"context"
-	"log"
 )
 
 type Locked[T any] struct {
@@ -26,9 +25,6 @@ func (s *Locked[T]) Modify(ctx context.Context, fn func(*T) error) error {
 	select {
 	case state := <-s.state:
 		defer func() { s.state <- state }()
-		log.Printf("locker %T to %p", state, fn)
-		defer log.Printf("locker %T from %p", state, fn)
-
 		return fn(state)
 	case <-ctx.Done():
 		return ctx.Err()

pkg/math/math.go

@@ -38,3 +38,30 @@ func Min[T ordered](i T, candidates ...T) T {
 	}
 	return i
 }
+
+func PagerBox(first, last uint64, pos, count int64) (uint64, int64) {
+	var start uint64
+
+	if pos >= 0 {
+		start = first + uint64(pos)
+	} else {
+		start = uint64(int64(last) + pos + 1)
+	}
+
+	switch {
+	case count > 0:
+		count = Min(count, int64(last-start)+1)
+
+	case pos >= 0 && count < 0:
+		count = Max(count, int64(first-start))
+
+	case pos < 0 && count < 0:
+		count = Max(count, int64(first-start)-1)
+	}
+
+	if count == 0 || (start < first && count <= 0) || (start > last && count >= 0) {
+		return 0, 0
+	}
+
+	return start, count
+}

pkg/math/math_test.go

@@ -1,9 +1,11 @@
 package math_test
 
 import (
+	"log"
 	"testing"
 
 	"github.com/matryer/is"
+	"github.com/sour-is/ev/pkg/es"
 	"github.com/sour-is/ev/pkg/math"
 )
 
@@ -46,3 +48,46 @@ func TestMath(t *testing.T) {
 	))
 
 }
+
+func TestPagerBox(t *testing.T) {
+	is := is.New(t)
+
+	tests := []struct {
+		first uint64
+		last  uint64
+		pos   int64
+		n     int64
+
+		start uint64
+		count int64
+	}{
+		{1, 10, 0, 10, 1, 10},
+		{1, 10, 0, 11, 1, 10},
+		{1, 5, 0, 10, 1, 5},
+		{1, 10, 4, 10, 5, 6},
+		{1, 10, 5, 10, 6, 5},
+		{1, 10, 0, -10, 0, 0},
+		{1, 10, 1, -1, 2, -1},
+		{1, 10, 1, -10, 2, -1},
+		{1, 10, -1, 1, 10, 1},
+		{1, 10, -2, 10, 9, 2},
+		{1, 10, -1, -1, 10, -1},
+		{1, 10, -2, -10, 9, -9},
+		{1, 10, 0, -10, 0, 0},
+		{1, 10, 10, 10, 0, 0},
+		{1, 10, 0, es.AllEvents, 1, 10},
+		{1, 10, -1, -es.AllEvents, 10, -10},
+	}
+
+	for _, tt := range tests {
+		start, count := math.PagerBox(tt.first, tt.last, tt.pos, tt.n)
+		if count > 0 {
+			log.Print(tt, "|", start, count, int64(start)+count-1)
+		} else {
+			log.Print(tt, "|", start, count, int64(start)+count+1)
+		}
+
+		is.Equal(start, tt.start)
+		is.Equal(count, tt.count)
+	}
+}