Merge pull request #8 from CastyLab/develop

Adding redis PubSub actions for user and theater events

config.example.yml

@@ -10,8 +10,9 @@ secrets:
     user: "service"
     pass: "super-secure-password"
   redis:
-    host: "localhost"
-    port: 6379
+    masterName: "mymaster"
+    sentinels:
+    - "localhost:6379"
     pass: "super-secure-password"
   oauth:
     google: "./config/oauth/google_client_secret.json"

config/config.go

@@ -27,9 +27,10 @@ type ConfMap struct {
 			Spotify string `yaml:"spotify"`
 		} `yaml:"oauth"`
 		Redis struct {
-			Host string `yaml:"host"`
-			Port int    `yaml:"port"`
-			Pass string `yaml:"pass"`
+			MasterName  string     `yaml:"masterName"`
+			Sentinels   []string   `yaml:"sentinels"`
+			Port        int        `yaml:"port"`
+			Pass        string     `yaml:"pass"`
 		} `yaml:"redis"`
 		JWT struct {
 			ExpireTime            int    `yaml:"expire_time"`

db/models/theater.go

@@ -28,4 +28,10 @@ type Follow struct {
 
 	CreatedAt          time.Time                  `bson:"created_at, omitempty" json:"created_at, omitempty"`
 	UpdatedAt          time.Time                  `bson:"updated_at, omitempty" json:"updated_at, omitempty"`
+}
+
+type TheaterMember struct {
+	ID                 *primitive.ObjectID        `bson:"_id, omitempty" json:"id, omitempty"`
+	TheaterId          *primitive.ObjectID        `bson:"theater_id, omitempty" json:"theater_id, omitempty"`
+	UserId             *primitive.ObjectID        `bson:"user_id, omitempty" json:"user_id, omitempty"`
 }

go.mod

@@ -3,17 +3,17 @@ module github.com/CastyLab/grpc.server
 go 1.14
 
 require (
-	github.com/CastyLab/grpc.proto v0.0.0-20200829020729-8a1f43db6b04
+	github.com/CastyLab/grpc.proto v0.0.0-20201024203352-28756769b0b7
 	github.com/asticode/go-astisub v0.2.0
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/getsentry/sentry-go v0.5.1
-	github.com/golang/protobuf v1.4.1
-	github.com/pingcap/errors v0.11.4
+	github.com/go-redis/redis/v8 v8.3.2
+	github.com/golang/protobuf v1.4.2
 	github.com/pkg/errors v0.8.1
 	go.mongodb.org/mongo-driver v1.3.1
-	golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
 	golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be
 	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013
 	google.golang.org/grpc v1.28.0
-	gopkg.in/yaml.v2 v2.2.4
+	gopkg.in/yaml.v2 v2.3.0
 )

helpers/member.go

@@ -0,0 +1,47 @@
+package helpers
+
+import (
+	"context"
+	"fmt"
+	"github.com/CastyLab/grpc.proto/proto"
+	"github.com/CastyLab/grpc.server/db"
+	"github.com/CastyLab/grpc.server/db/models"
+	"go.mongodb.org/mongo-driver/bson"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func NewMemberProto(member *models.TheaterMember) (*proto.User, error) {
+	var (
+		dbmember = new(models.User)
+		decoder  = db.Connection.Collection("users").
+			FindOne(context.Background(), bson.M{"_id": member.UserId})
+	)
+	if err := decoder.Decode(dbmember); err != nil {
+		return nil, fmt.Errorf("could not decode theater member: %v", err)
+	}
+	return NewProtoUser(dbmember), nil
+}
+
+func GetTheaterMembers(ctx context.Context, theater *models.Theater) ([]*proto.User, error) {
+
+	members := make([]*proto.User, 0)
+	cursor, err := db.Connection.Collection("theater_members").Find(ctx, bson.M{"theater_id": theater.ID})
+	if err != nil {
+		return nil, status.Error(codes.Internal, "Could not get theater members")
+	}
+
+	for cursor.Next(ctx) {
+		member := new(models.TheaterMember)
+		if err := cursor.Decode(member); err != nil {
+			continue
+		}
+		protoMember, err := NewMemberProto(member)
+		if err != nil {
+			continue
+		}
+		members = append(members, protoMember)
+	}
+
+	return members, nil
+}

helpers/message.go

@@ -12,6 +12,7 @@ import (
 func NewProtoMessage(ctx context.Context, message *models.Message) (*proto.Message, error) {
 
 	var (
+		err error
 		dbSender   = new(models.User)
 		collection = db.Connection.Collection("users")
 	)
@@ -20,11 +21,7 @@ func NewProtoMessage(ctx context.Context, message *models.Message) (*proto.Messa
 		return nil, err
 	}
 
-	sender, err := NewProtoUser(dbSender)
-	if err != nil {
-		return nil, err
-	}
-
+	sender := NewProtoUser(dbSender)
 	createdAt, _ := ptypes.TimestampProto(message.CreatedAt)
 	updatedAt, _ := ptypes.TimestampProto(message.UpdatedAt)
 

helpers/notifications.go

@@ -30,17 +30,12 @@ func NewNotificationProto(notif *models.Notification) (*proto.Notification, erro
 		return nil, err
 	}
 
-	protoUser, err := NewProtoUser(fromUser)
-	if err != nil {
-		return nil, err
-	}
-
-	protoMSG := &proto.Notification{
+	protoMSG  := &proto.Notification{
 		Id:        notif.ID.Hex(),
 		Type:      notif.Type,
 		Read:      notif.Read,
 		ReadAt:    readAt,
-		FromUser:  protoUser,
+		FromUser:  NewProtoUser(fromUser),
 		CreatedAt: createdAt,
 		UpdatedAt: updatedAt,
 	}

helpers/theater.go

@@ -12,7 +12,6 @@ import (
 func NewTheaterProto(ctx context.Context, theater *models.Theater) (*proto.Theater, error) {
 
 	var (
-		err error
 		database                 = db.Connection
 		thUser                   = new(models.User)
 		mediaSourceProtoMessage  = new(proto.MediaSource)
@@ -33,15 +32,10 @@ func NewTheaterProto(ctx context.Context, theater *models.Theater) (*proto.Theat
 		return nil, err
 	}
 
-	thProtoMessageUser, err := NewProtoUser(thUser)
-	if err != nil {
-		return nil, err
-	}
-
 	return &proto.Theater{
 		Id:                theater.ID.Hex(),
 		Description:       theater.Description,
-		User:              thProtoMessageUser,
+		User:              NewProtoUser(thUser),
 		MediaSource:       mediaSourceProtoMessage,
 		Privacy:           theater.Privacy,
 		VideoPlayerAccess: theater.VideoPlayerAccess,

helpers/user.go

@@ -1,18 +1,90 @@
 package helpers
 
 import (
+	"context"
+	"fmt"
 	"github.com/CastyLab/grpc.proto/proto"
+	"github.com/CastyLab/grpc.server/db"
 	"github.com/CastyLab/grpc.server/db/models"
+	"github.com/CastyLab/grpc.server/redis"
 	"github.com/golang/protobuf/ptypes"
+	"go.mongodb.org/mongo-driver/bson"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
-func NewProtoUser(user *models.User) (*proto.User, error) {
+func GetFriendsFromDatabase(ctx context.Context, user *models.User) ([]*proto.User, error) {
+	var (
+		friends           = make([]*proto.User, 0)
+		userCollection    = db.Connection.Collection("users")
+		friendsCollection = db.Connection.Collection("friends")
+	)
 
+	filter := bson.M{
+		"accepted": true,
+		"$or": []interface{}{
+			bson.M{"friend_id": user.ID},
+			bson.M{"user_id": user.ID},
+		},
+	}
+
+	cursor, err := friendsCollection.Find(ctx, filter)
+	if err != nil {
+		return nil, status.Error(codes.NotFound, "Could not find friends!")
+	}
+
+	for cursor.Next(ctx) {
+
+		var friend = new(models.Friend)
+		if err := cursor.Decode(friend); err != nil {
+			continue
+		}
+
+		var filter = bson.M{"_id": friend.FriendId}
+		if user.ID.Hex() == friend.FriendId.Hex() {
+			filter = bson.M{"_id": friend.UserId}
+		}
+
+		friendUserObject := new(models.User)
+		if err := userCollection.FindOne(ctx, filter).Decode(friendUserObject); err != nil {
+			continue
+		}
+
+		friends = append(friends, NewProtoUser(friendUserObject))
+	}
+
+	return friends, nil
+}
+
+// update friends with new event of user
+func SendEventToFriends(ctx context.Context, event []byte, user *models.User) error {
+	friends, err := GetFriendsFromDatabase(ctx, user)
+	if err != nil {
+		return status.Error(codes.Internal, "Could not get friends!")
+	}
+	SendEventToUsers(ctx, event, friends)
+	return nil
+}
+
+func SendEventToUser(ctx context.Context, event []byte, user *proto.User)  {
+	redis.Client.Publish(ctx, fmt.Sprintf("user:events:%s", user.Id), event)
+}
+
+func SendEventToUsers(ctx context.Context, event []byte, users []*proto.User)  {
+	for _, user := range users {
+		redis.Client.Publish(ctx, fmt.Sprintf("user:events:%s", user.Id), event)
+	}
+}
+
+func SendEventToTheaterMembers(ctx context.Context, event []byte, theater *models.Theater)  {
+	redis.Client.Publish(ctx, fmt.Sprintf("theater:events:%s", theater.ID.Hex()), event)
+}
+
+func NewProtoUser(user *models.User) *proto.User {
 	lastLogin, _ := ptypes.TimestampProto(user.LastLogin)
 	joinedAt,  _ := ptypes.TimestampProto(user.JoinedAt)
 	updatedAt, _ := ptypes.TimestampProto(user.UpdatedAt)
-
-	protoUser := &proto.User{
+	return &proto.User{
 		Id:             user.ID.Hex(),
 		Fullname:       user.Fullname,
 		Username:       user.Username,
@@ -29,6 +101,4 @@ func NewProtoUser(user *models.User) (*proto.User, error) {
 		JoinedAt:       joinedAt,
 		UpdatedAt:      updatedAt,
 	}
-
-	return protoUser, nil
 }

oauth/google/google.go

@@ -34,6 +34,7 @@ func Configure() error {
 }
 
 func Authenticate(code string) (*oauth2.Token, error) {
-	mCtx, _ := context.WithTimeout(context.Background(), 10 * time.Second)
+	mCtx, cancel := context.WithTimeout(context.Background(), 10 * time.Second)
+	defer cancel()
 	return oauthClient.Exchange(mCtx, code, oauth2.AccessTypeOffline)
 }

redis/redis.go

@@ -0,0 +1,31 @@
+package redis
+
+import (
+	"context"
+	"github.com/CastyLab/grpc.server/config"
+	"github.com/go-redis/redis/v8"
+	"log"
+)
+
+var (
+	Client *redis.Client
+)
+
+func Configure() error {
+	Client = redis.NewFailoverClient(&redis.FailoverOptions{
+		SentinelAddrs: config.Map.Secrets.Redis.Sentinels,
+		Password: config.Map.Secrets.Redis.Pass,
+		MasterName: config.Map.Secrets.Redis.MasterName,
+		DB: 0,
+	})
+	cmd := Client.Ping(context.Background())
+	if res := cmd.Val(); res != "PONG" {
+		log.Println("SentinelAddrs: ", config.Map.Secrets.Redis.Sentinels)
+		log.Fatalf("Could not ping the redis server: %v", cmd.Err())
+	}
+	return nil
+}
+
+func Close() error {
+	return Client.Close()
+}

server.go

@@ -8,6 +8,7 @@ import (
 	"github.com/CastyLab/grpc.server/db"
 	"github.com/CastyLab/grpc.server/jwt"
 	"github.com/CastyLab/grpc.server/oauth"
+	"github.com/CastyLab/grpc.server/redis"
 	"github.com/CastyLab/grpc.server/services/auth"
 	"github.com/CastyLab/grpc.server/services/message"
 	"github.com/CastyLab/grpc.server/services/theater"
@@ -42,6 +43,10 @@ func init() {
 		log.Fatal(fmt.Errorf("could not load config: %v", err))
 	}
 
+	if err := redis.Configure(); err != nil {
+		log.Fatal(fmt.Errorf("could not configure redis : %v", err))
+	}
+
 	if err := sentry.Init(sentry.ClientOptions{ Dsn: config.Map.Secrets.SentryDsn }); err != nil {
 		log.Fatal(fmt.Errorf("could not initilize sentry: %v", err))
 	}
@@ -68,9 +73,17 @@ func init() {
 
 func main() {
 
-	// Since sentry emits events in the background we need to make sure
-	// they are sent before we shut down
-	defer sentry.Flush(time.Second * 5)
+	defer func() {
+
+		// Since sentry emits events in the background we need to make sure
+		// they are sent before we shut down
+		sentry.Flush(time.Second * 5)
+
+		if err := redis.Close(); err != nil {
+			log.Println(fmt.Errorf("could not close redis connection: %v", err))
+		}
+
+	}()
 
 	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", *host, *port))
 	if err != nil {

services/auth/service.go

@@ -8,17 +8,14 @@ import (
 	"github.com/CastyLab/grpc.server/jwt"
 	"github.com/getsentry/sentry-go"
 	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/mongo"
 	"golang.org/x/crypto/bcrypt"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"net/http"
 	"regexp"
 )
 
-type Service struct {
-	db *mongo.Database
-}
+type Service struct {}
 
 func (s *Service) isEmail(user string) bool {