rewind feature to stream method

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.13.2
+current_version = 0.13.3
 commit = False
 tag = False
 

CHANGELOG.md

@@ -60,3 +60,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [0.12.1] - 2020-12-23
 - PutDoc, PutDocs, PutConnection, PutConnections for full create-or-replace functionality
+
+## [0.13.2] - 2020-12-28
+- Add "rewind" feature to all stream method to capture historical messages

Makefile

@@ -1,4 +1,4 @@
-version := "0.13.2"
+version := "0.13.3"
 
 .DEFAULT_GOAL := help
 

README.md

@@ -7,7 +7,7 @@ https://graphikdb.github.io/graphik/
 
 `git clone git@github.com:graphikDB/graphik.git`
 
-`docker pull graphikdb/graphik:v0.13.2`
+`docker pull graphikdb/graphik:v0.13.3`
 
 Graphik is a Backend as a Service implemented as an identity-aware, permissioned, persistant document/graph database & pubsub server written in Go.
 
@@ -910,7 +910,7 @@ add this docker-compose.yml to ${pwd}:
     version: '3.7'
     services:
       graphik:
-        image: graphikdb/graphik:v0.13.2
+        image: graphikdb/graphik:v0.13.3
         env_file:
           - .env
         ports:

database/constants.go

@@ -28,6 +28,7 @@ var (
 	dbConstraints      = []byte("constraints")
 	dbIndexDocs        = []byte("indexedDocs")
 	dbIndexConnections = []byte("indexedConnections")
+	dbMessages         = []byte("messages")
 	// An error indicating a given key does not exist
 	ErrNotFound             = errors.New("not found")
 	ErrAlreadyExists        = errors.New("already exists")

database/graph.go

@@ -1,6 +1,7 @@
 package database
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -23,6 +24,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/structpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"io/ioutil"
@@ -39,11 +41,11 @@ import (
 type Graph struct {
 	// db is the underlying handle to the db.
 	db              *bbolt.DB
+	pubsub          *bbolt.DB
 	jwksMu          sync.RWMutex
 	jwksSet         *jwk.Set
 	jwtCache        *generic.Cache
 	openID          *openIDConnect
-	path            string
 	mu              sync.RWMutex
 	connectionsTo   map[string]map[string]struct{}
 	connectionsFrom map[string]map[string]struct{}
@@ -63,21 +65,24 @@ type Graph struct {
 // NewGraph takes a file path and returns a connected Raft backend.
 func NewGraph(ctx context.Context, flgs *apipb.Flags, lgger *logger.Logger) (*Graph, error) {
 	os.MkdirAll(flgs.StoragePath, 0700)
-	path := filepath.Join(flgs.StoragePath, "graph.db")
-	handle, err := bbolt.Open(path, dbFileMode, nil)
+	graphDB, err := bbolt.Open(filepath.Join(flgs.StoragePath, "graph.db"), dbFileMode, nil)
+	if err != nil {
+		return nil, err
+	}
+	pubsubDB, err := bbolt.Open(filepath.Join(flgs.StoragePath, "pubsub.db"), dbFileMode, nil)
 	if err != nil {
 		return nil, err
 	}
 
 	var closers []func()
 	m := machine.New(ctx, machine.WithMaxRoutines(100000))
 	g := &Graph{
-		db:              handle,
+		db:              graphDB,
 		jwksMu:          sync.RWMutex{},
 		jwksSet:         nil,
 		jwtCache:        generic.NewCache(5 * time.Minute),
 		openID:          nil,
-		path:            path,
+		pubsub:          pubsubDB,
 		mu:              sync.RWMutex{},
 		connectionsTo:   map[string]map[string]struct{}{},
 		connectionsFrom: map[string]map[string]struct{}{},
@@ -153,6 +158,16 @@ func NewGraph(ctx context.Context, flgs *apipb.Flags, lgger *logger.Logger) (*Gr
 	if err != nil {
 		return nil, err
 	}
+	if err := pubsubDB.Update(func(tx *bbolt.Tx) error {
+		_, err = tx.CreateBucketIfNotExists(dbMessages)
+		if err != nil {
+			return errors.Wrap(err, "failed to create messages bucket")
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+
 	if err := g.cacheConnectionRefs(); err != nil {
 		return nil, err
 	}
@@ -860,7 +875,7 @@ func (g *Graph) Broadcast(ctx context.Context, message *apipb.OutboundMessage) (
 		return nil, prepareError(ErrFailedToGetUser)
 	}
 	if message.GetChannel() == changeChannel {
-		return nil, status.Error(codes.PermissionDenied, "forbidden from publishing to the changes channel")
+		return nil, status.Error(codes.PermissionDenied, "forbidden from broadcasting to the state channel")
 	}
 	_, err := g.applyCommand(&apipb.RaftCommand{
 		User:   user,
@@ -912,6 +927,41 @@ func (g *Graph) Stream(filter *apipb.StreamFilter, server apipb.DatabaseService_
 			return true
 		}
 	}
+	if filter.GetRewind() != "" {
+		dur, err := time.ParseDuration(filter.GetRewind())
+		if err != nil {
+			return status.Error(codes.InvalidArgument, err.Error())
+		}
+		if err := g.pubsub.View(func(tx *bbolt.Tx) error {
+			msgsBucket := tx.Bucket(dbMessages)
+			bucket := msgsBucket.Bucket([]byte(filter.Channel))
+			if bucket == nil {
+				bucket, err = msgsBucket.CreateBucketIfNotExists([]byte(filter.Channel))
+				if err != nil {
+					return err
+				}
+			}
+			c := bucket.Cursor()
+
+			min := helpers.Uint64ToBytes(uint64(time.Now().Truncate(dur).UnixNano()))
+			max := helpers.Uint64ToBytes(uint64(time.Now().UnixNano()))
+			for k, v := c.Seek(min); k != nil && bytes.Compare(k, max) <= 0; k, v = c.Next() {
+				var msg = &apipb.Message{}
+				if err := proto.Unmarshal(v, msg); err != nil {
+					return errors.Wrap(err, "failed to unmarshal message")
+				}
+				if filterFunc(msg) {
+					if err := server.Send(msg); err != nil {
+						return err
+					}
+				}
+			}
+			return nil
+		}); err != nil {
+			return prepareError(err)
+		}
+	}
+
 	if err := g.machine.PubSub().SubscribeFilter(server.Context(), filter.GetChannel(), filterFunc, func(msg interface{}) {
 		if err, ok := msg.(error); ok && err != nil {
 			g.logger.Error("failed to send subscription", zap.Error(err))
@@ -939,7 +989,10 @@ func (b *Graph) Close() {
 		}
 		b.machine.Wait()
 		if err := b.db.Close(); err != nil {
-			b.logger.Error("failed to close db", zap.Error(err))
+			b.logger.Error("failed to close graph db", zap.Error(err))
+		}
+		if err := b.pubsub.Close(); err != nil {
+			b.logger.Error("failed to close pubsub db", zap.Error(err))
 		}
 	})
 }

database/raft.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	apipb "github.com/graphikDB/graphik/gen/grpc/go"
+	"github.com/graphikDB/graphik/helpers"
 	"github.com/graphikDB/raft/fsm"
 	"github.com/hashicorp/raft"
 	"github.com/pkg/errors"
@@ -128,6 +129,26 @@ func (g *Graph) fsm() *fsm.FSM {
 				}
 			}
 			if cmd.GetSendMessage() != nil {
+				if err := g.pubsub.Update(func(tx *bbolt.Tx) error {
+					msgsBucket := tx.Bucket(dbMessages)
+					bucket := msgsBucket.Bucket([]byte(cmd.SendMessage.Channel))
+					if bucket == nil {
+						bucket, err = msgsBucket.CreateBucketIfNotExists([]byte(cmd.SendMessage.Channel))
+						if err != nil {
+							return err
+						}
+					}
+					bits, err := proto.Marshal(cmd.SendMessage)
+					if err != nil {
+						return err
+					}
+					if err := bucket.Put(helpers.Uint64ToBytes(uint64(cmd.SendMessage.Timestamp.AsTime().UnixNano())), bits); err != nil {
+						return err
+					}
+					return nil
+				}); err != nil {
+					return err
+				}
 				if err := g.machine.PubSub().Publish(cmd.SendMessage.Channel, cmd.SendMessage); err != nil {
 					return status.Error(codes.Internal, err.Error())
 				}

docker-compose.yml

@@ -1,7 +1,7 @@
 version: '3.7'
 services:
   graphik:
-    image: graphikdb/graphik:v0.13.2
+    image: graphikdb/graphik:v0.13.3
     env_file:
       - .env
     ports:

gen/gql/docs/streamfilter.doc.html

@@ -464,7 +464,7 @@ <h2 id="graphql-schema-definition" class="graphdoc-section__title slds-text-head
             </a>
             GraphQL Schema definition
         </h2>
-        <code class="highlight"><ul class="code" style="padding-left:28px"><li><span class="keyword operator ts">input</span> <span class="identifier">StreamFilter</span> {</li><li><span class="tab"><li><span class="tab"><span class="comment line">#   channel is the target channel to listen on</span></span></li><li><span class="tab"><span class="meta">channel</span>: <a class="support type" href="string.doc.html">String</a>!</span></li></span></li><li><span class="tab"><li><span class="tab"><span class="comment line">#   expression is a CEL expression used to filter messages</span></span></li><li><span class="tab"><span class="meta">expression</span>: <a class="support type" href="string.doc.html">String</a></span></li></span></li><li>}</li></ul></code>
+        <code class="highlight"><ul class="code" style="padding-left:28px"><li><span class="keyword operator ts">input</span> <span class="identifier">StreamFilter</span> {</li><li><span class="tab"><li><span class="tab"><span class="comment line">#   channel is the target channel to listen on</span></span></li><li><span class="tab"><span class="meta">channel</span>: <a class="support type" href="string.doc.html">String</a>!</span></li></span></li><li><span class="tab"><li><span class="tab"><span class="comment line">#   expression is a CEL expression used to filter messages</span></span></li><li><span class="tab"><span class="meta">expression</span>: <a class="support type" href="string.doc.html">String</a></span></li></span></li><li><span class="tab"><li><span class="tab"><span class="comment line">#   rewind time by a specified duration to capture messages from history</span></span></li><li><span class="tab"><span class="meta">rewind</span>: <a class="support type" href="string.doc.html">String</a></span></li></span></li><li>}</li></ul></code>
     </div>
 </section>
 <section>