Implement events streaming

main.go

@@ -5,7 +5,6 @@ import (
 	"net"
 
 	"github.com/breez/data-sync/config"
-	"github.com/breez/data-sync/middleware"
 	"github.com/breez/data-sync/proto"
 	"google.golang.org/grpc"
 )
@@ -20,16 +19,17 @@ func main() {
 		log.Fatalf("Failed to listen: %v", err)
 	}
 
-	s := CreateServer(config, grpcListener)
+	quitChan := make(chan struct{})
+	syncServer := NewPersistentSyncerServer(config)
+	syncServer.Start(quitChan)
+	s := CreateServer(config, grpcListener, syncServer)
 	if err := s.Serve(grpcListener); err != nil {
 		log.Fatalf("failed to serve: %v", err)
 	}
 }
 
-func CreateServer(config *config.Config, listener net.Listener) *grpc.Server {
-	s := grpc.NewServer(grpc.ChainUnaryInterceptor(middleware.UnaryAuth(config)))
-	proto.RegisterSyncerServer(s, &PersistentSyncerServer{
-		config: config,
-	})
+func CreateServer(config *config.Config, listener net.Listener, syncServer proto.SyncerServer) *grpc.Server {
+	s := grpc.NewServer()
+	proto.RegisterSyncerServer(s, syncServer)
 	return s
 }

middleware/auth.go

@@ -14,53 +14,57 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2/ecdsa"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/tv42/zbase32"
-
-	"google.golang.org/grpc"
 )
 
 const (
-	USER_DB_CONTEXT_KEY = "user_db"
+	USER_DB_CONTEXT_KEY     = "user_db"
+	USER_PUBKEY_CONTEXT_KEY = "user_pubkey"
 )
 
 var ErrInternalError = fmt.Errorf("internal error")
 var ErrInvalidSignature = fmt.Errorf("invalid signature")
 var SignedMsgPrefix = []byte("Lightning Signed Message:")
 
-func UnaryAuth(config *config.Config) grpc.UnaryServerInterceptor {
-	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+func Authenticate(config *config.Config, ctx context.Context, req interface{}) (context.Context, error) {
+	var toVerify string
+	var signature string
+	setRecordReq, ok := req.(*proto.SetRecordRequest)
+	if ok {
+		toVerify = fmt.Sprintf("%v-%v-%x-%v", setRecordReq.Record.Id, setRecordReq.Record.Version, setRecordReq.Record.Data, setRecordReq.RequestTime)
+		signature = setRecordReq.Signature
+	}
 
-		var toVerify string
-		var signature string
-		setRecordReq, ok := req.(*proto.SetRecordRequest)
-		if ok {
-			toVerify = fmt.Sprintf("%v-%v-%x-%v", setRecordReq.Record.Id, setRecordReq.Record.Version, setRecordReq.Record.Data, setRecordReq.RequestTime)
-			signature = setRecordReq.Signature
-		}
+	listChangesReq, ok := req.(*proto.ListChangesRequest)
+	if ok {
+		toVerify = fmt.Sprintf("%v-%v", listChangesReq.SinceVersion, listChangesReq.RequestTime)
+		signature = listChangesReq.Signature
+	}
 
-		listChangesReq, ok := req.(*proto.ListChangesRequest)
-		if ok {
-			toVerify = fmt.Sprintf("%v-%v", listChangesReq.SinceVersion, listChangesReq.RequestTime)
-			signature = listChangesReq.Signature
-		}
+	trackChangesReq, ok := req.(*proto.TrackChangesRequest)
+	if ok {
+		toVerify = fmt.Sprintf("%v-%v", trackChangesReq.SinceVersion, trackChangesReq.RequestTime)
+		signature = trackChangesReq.Signature
+	}
 
-		pubkey, err := VerifyMessage([]byte(toVerify), signature)
-		if err != nil {
-			return nil, err
-		}
+	pubkey, err := VerifyMessage([]byte(toVerify), signature)
+	if err != nil {
+		return nil, err
+	}
 
-		dbDir := config.UsersDatabasesDir
-		pubkeyBytes := pubkey.SerializeCompressed()
-		storeFile := fmt.Sprintf("%v/%v/%v/%v", dbDir,
-			hex.EncodeToString(pubkeyBytes[0:1]),
-			hex.EncodeToString(pubkeyBytes[1:2]),
-			hex.EncodeToString(pubkeyBytes[2:]))
-		db, err := store.Connect(storeFile)
-		if err != nil {
-			log.Printf("failed to connect to database file %v: %v", storeFile, err)
-			return nil, ErrInternalError
-		}
-		return handler(context.WithValue(ctx, USER_DB_CONTEXT_KEY, db), req)
+	dbDir := config.UsersDatabasesDir
+	pubkeyBytes := pubkey.SerializeCompressed()
+	storeFile := fmt.Sprintf("%v/%v/%v/%v", dbDir,
+		hex.EncodeToString(pubkeyBytes[0:1]),
+		hex.EncodeToString(pubkeyBytes[1:2]),
+		hex.EncodeToString(pubkeyBytes[2:]))
+	db, err := store.Connect(storeFile)
+	if err != nil {
+		log.Printf("failed to connect to database file %v: %v", storeFile, err)
+		return nil, ErrInternalError
 	}
+	newContext := context.WithValue(ctx, USER_DB_CONTEXT_KEY, db)
+	newContext = context.WithValue(newContext, USER_PUBKEY_CONTEXT_KEY, hex.EncodeToString(pubkeyBytes))
+	return newContext, nil
 }
 
 func SignMessage(key *btcec.PrivateKey, msg []byte) (string, error) {

proto/sync.proto

@@ -6,6 +6,7 @@ package proto;
 service Syncer {
   rpc SetRecord(SetRecordRequest) returns (SetRecordReply) {}
   rpc ListChanges(ListChangesRequest) returns (ListChangesReply) {}
+  rpc TrackChanges(TrackChangesRequest) returns (stream Record);
 }
 
 message Record {
@@ -34,3 +35,9 @@ message ListChangesRequest {
   string signature = 3;
 }
 message ListChangesReply { repeated Record changes = 1; }
+
+message TrackChangesRequest {
+  int64 since_version = 1;
+  int64 request_time = 2;
+  string signature = 3;
+}