Adding MediaSourceChange event and removing upload service since we h…

…ave api server

db/models/media_source.go

@@ -16,7 +16,6 @@ type MediaSource struct {
 	Banner          string                    `bson:"banner" json:"banner,omitempty"`
 	Uri             string                    `bson:"uri" json:"uri,omitempty"`
 
-	LastPlayedTime  int64                     `bson:"last_played_time" json:"last_played_time,omitempty"`
 	Length          int64                     `bson:"length" json:"length,omitempty"`
 
 	Subtitles       []*Subtitle               `json:"subtitles,omitempty"`

go.mod

@@ -3,7 +3,7 @@ module github.com/CastyLab/grpc.server
 go 1.14
 
 require (
-	github.com/CastyLab/grpc.proto v0.0.0-20200817114812-aded2da55bf8
+	github.com/CastyLab/grpc.proto v0.0.0-20200817142707-b9a8e630b5a4
 	github.com/MrJoshLab/go-respond v0.0.0-20191125172458-25d4d18aaa1b
 	github.com/asticode/go-astisub v0.2.0
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible

go.sum

@@ -140,6 +140,10 @@ github.com/CastyLab/grpc.proto v0.0.0-20200817113914-fa2f219ad733 h1:nfqIDh8JTQ9
 github.com/CastyLab/grpc.proto v0.0.0-20200817113914-fa2f219ad733/go.mod h1:5ypgijUv7rWXMsRbUZXEgTucAA+4BpHtXgGIvOyUGRU=
 github.com/CastyLab/grpc.proto v0.0.0-20200817114812-aded2da55bf8 h1:Rzx+c/bDPjvTWSO2bvelp8mGZ+Y1UWxl1uZoMP5ebFk=
 github.com/CastyLab/grpc.proto v0.0.0-20200817114812-aded2da55bf8/go.mod h1:5ypgijUv7rWXMsRbUZXEgTucAA+4BpHtXgGIvOyUGRU=
+github.com/CastyLab/grpc.proto v0.0.0-20200817141530-d78b7c63ead8 h1:r4LZRlzIDbd4D6dDaae6hnii8jS9jFYekNPTsP4PoTA=
+github.com/CastyLab/grpc.proto v0.0.0-20200817141530-d78b7c63ead8/go.mod h1:5ypgijUv7rWXMsRbUZXEgTucAA+4BpHtXgGIvOyUGRU=
+github.com/CastyLab/grpc.proto v0.0.0-20200817142707-b9a8e630b5a4 h1:FjUniPgItkpqUeDVDDckr5iQ9xPcOsZ4W4V2RpLeefc=
+github.com/CastyLab/grpc.proto v0.0.0-20200817142707-b9a8e630b5a4/go.mod h1:5ypgijUv7rWXMsRbUZXEgTucAA+4BpHtXgGIvOyUGRU=
 github.com/CloudyKit/fastprinter v0.0.0-20170127035650-74b38d55f37a/go.mod h1:EFZQ978U7x8IRnstaskI3IysnWY5Ao3QgZUKOXlsAdw=
 github.com/CloudyKit/jet v2.1.3-0.20180809161101-62edd43e4f88+incompatible/go.mod h1:HPYO+50pSWkPoj9Q/eq0aRGByCL6ScRlUmiEX5Zgm+w=
 github.com/Joker/hpp v1.0.0/go.mod h1:8x5n+M1Hp5hC0g8okX3sR3vFQwynaX/UgSOM9MeBKzY=

helpers/theater.go

@@ -66,7 +66,6 @@ func NewMediaSourceProto(ms *models.MediaSource) (*proto.MediaSource, error) {
 		Type:             ms.Type,
 		Banner:           ms.Banner,
 		Uri:              ms.Uri,
-		LastPlayedTime:   ms.LastPlayedTime,
 		Length:           ms.Length,
 		Subtitles:        make([]*proto.Subtitle, 0),
 		CreatedAt:        createdAt,

internal/services/theater/update.go

@@ -12,6 +12,35 @@ type InternalWsTheaterService struct {
 	HttpClient http.Client
 }
 
+func (s *InternalWsTheaterService) SendMediaSourceUpdateEvent(theaterId, mediaSourceId string) error {
+	params := url.Values{}
+	params.Set("theater_id", theaterId)
+	params.Set("media_source_id", mediaSourceId)
+
+	request, err := http.NewRequest("POST", "http://unix/media_source/@updated", strings.NewReader(params.Encode()))
+	if err != nil {
+		return err
+	}
+
+	request.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	response, err := s.HttpClient.Do(request)
+	if err != nil {
+		return err
+	}
+
+	result := map[string] interface{}{}
+	if err := json.NewDecoder(response.Body).Decode(&result); err != nil {
+		return err
+	}
+
+	if result["status"] == "success" {
+		return nil
+	}
+
+	return errors.New("Something went wrong, Could not send event!")
+}
+
 func (s *InternalWsTheaterService) SendTheaterUpdateEvent(theaterId string) error {
 	params := url.Values{}
 	params.Set("theater_id", theaterId)

services/auth/service.go

@@ -8,14 +8,17 @@ 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 {}
+type Service struct {
+	db *mongo.Database
+}
 
 func (s *Service) isEmail(user string) bool {
 
@@ -43,17 +46,6 @@ func (s *Service) Authenticate(ctx context.Context, req *proto.AuthRequest) (*pr
 		unauthorized = status.Error(codes.Unauthenticated, "Unauthorized!")
 	)
 
-	//md, ok := metadata.FromIncomingContext(ctx)
-	//if !ok {
-	//	return nil, status.Error(codes.InvalidArgument, "Captcha is required!")
-	//}
-	//
-	//recaptcha := md.Get("g-recaptcha-response")
-	//if success, err := helpers.VerifyRecaptcha(recaptcha[0]); err != nil || !success {
-	//	log.Println(req, err)
-	//	return nil, status.Error(codes.InvalidArgument, "Captcha is required!")
-	//}
-
 	if req.User == "" || req.Pass == "" {
 		return nil, unauthorized
 	}

services/message/service.go

@@ -9,12 +9,15 @@ import (
 	"github.com/CastyLab/grpc.server/helpers"
 	"github.com/CastyLab/grpc.server/services/auth"
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"net/http"
 )
 
-type Service struct {}
+type Service struct {
+	db *mongo.Database
+}
 
 func (s *Service) GetUserMessages(ctx context.Context, req *proto.GetMessagesRequest) (*proto.GetMessagesResponse, error) {
 

services/theater/media_source.go

@@ -7,14 +7,19 @@ import (
 	"github.com/CastyLab/grpc.server/db"
 	"github.com/CastyLab/grpc.server/db/models"
 	"github.com/CastyLab/grpc.server/helpers"
+	"github.com/CastyLab/grpc.server/internal"
+	"github.com/CastyLab/grpc.server/services"
 	"github.com/CastyLab/grpc.server/services/auth"
+	"github.com/getsentry/sentry-go"
 	"github.com/golang/protobuf/ptypes/any"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"io"
 	"net/http"
+	"os"
 	"time"
 )
 
@@ -46,14 +51,24 @@ func (s *Service) SelectMediaSource(ctx context.Context, req *proto.MediaSourceA
 	}
 
 	if count != 0 {
-		_, err := collection.UpdateOne(ctx, bson.M{ "user_id": user.ID }, bson.M{
+
+		update, err := collection.UpdateOne(ctx, bson.M{ "user_id": user.ID }, bson.M{
 			"$set": bson.M{
 				"media_source_id": mediaSourceId,
 			},
 		})
 		if err != nil {
 			return nil, err
 		}
+
+		theaterId := update.UpsertedID.(primitive.ObjectID)
+
+		// sending new media source to websocket
+		err = internal.Client.TheaterService.SendMediaSourceUpdateEvent(mediaSourceId.Hex(), theaterId.Hex())
+		if err != nil {
+			sentry.CaptureException(err)
+		}
+
 		return &proto.Response{
 			Status:  "success",
 			Code:    http.StatusOK,
@@ -64,6 +79,27 @@ func (s *Service) SelectMediaSource(ctx context.Context, req *proto.MediaSourceA
 	return nil, emptyResponse
 }
 
+func (s *Service) SavePosterFromUrl(url string) (string, error) {
+	var (
+		storagePath = os.Getenv("STORAGE_PATH")
+		posterName  = services.RandomNumber(20)
+	)
+	avatarFile, err := os.Create(fmt.Sprintf("%s/uploads/posters/%s.png", storagePath, posterName))
+	if err != nil {
+		return posterName, err
+	}
+	defer avatarFile.Close()
+	resp, err := http.DefaultClient.Get(url)
+	if err != nil {
+		return posterName, err
+	}
+	defer resp.Body.Close()
+	if _, err := io.Copy(avatarFile, resp.Body); err != nil {
+		return posterName, err
+	}
+	return posterName, nil
+}
+
 func (s *Service) AddMediaSource(ctx context.Context, req *proto.MediaSourceAuthRequest) (*proto.Response, error) {
 
 	var (
@@ -101,12 +137,18 @@ func (s *Service) AddMediaSource(ctx context.Context, req *proto.MediaSourceAuth
 		})
 	}
 
+	var poster string
+	poster, err = s.SavePosterFromUrl(req.Media.Banner)
+	if err != nil {
+		sentry.CaptureException(fmt.Errorf("could not upload poster %v", err))
+		poster = "default"
+	}
+
 	mediaSource := bson.M{
 		"title":              req.Media.Title,
 		"type":               req.Media.Type,
-		"banner":             req.Media.Banner,
+		"banner":             poster,
 		"uri":                req.Media.Uri,
-		"last_played_time":   req.Media.LastPlayedTime,
 		"length":             req.Media.Length,
 		"user_id":            user.ID,
 		"created_at":         time.Now(),
@@ -118,15 +160,20 @@ func (s *Service) AddMediaSource(ctx context.Context, req *proto.MediaSourceAuth
 		return nil, failedResponse
 	}
 
-	insertedID := result.InsertedID.(primitive.ObjectID)
-	uResult, _ := theatersCollection.UpdateOne(ctx, bson.M{"user_id": user.ID}, bson.M{
+	var (
+		theaterId primitive.ObjectID
+		insertedID = result.InsertedID.(primitive.ObjectID)
+	)
+
+	update, _ := theatersCollection.UpdateOne(ctx, bson.M{"user_id": user.ID}, bson.M{
 		"$set": bson.M{
 			"media_source_id": insertedID,
 		},
 	})
 
-	if uResult.ModifiedCount == 0 {
-		_, _ = theatersCollection.InsertOne(ctx, bson.M{
+	// if theater didnt match, then we will create one for user
+	if update.MatchedCount == 0 {
+		insertedTheater, err := theatersCollection.InsertOne(ctx, bson.M{
 			"title":               fmt.Sprintf("%s's Theater", user.Fullname),
 			"privacy":             proto.PRIVACY_PUBLIC,
 			"video_player_access": proto.VIDEO_PLAYER_ACCESS_ACCESS_BY_USER,
@@ -135,6 +182,21 @@ func (s *Service) AddMediaSource(ctx context.Context, req *proto.MediaSourceAuth
 			"created_at":          time.Now(),
 			"updated_at":          time.Now(),
 		})
+		if err != nil {
+			if _, err := collection.DeleteOne(ctx, bson.M{ "_id": insertedID }); err != nil {
+				sentry.CaptureException(fmt.Errorf("could not delete media source while creating new theater %v", err))
+			}
+			return nil, status.Errorf(codes.Internal, "could not create media source, please try again later!")
+		}
+		theaterId = insertedTheater.InsertedID.(primitive.ObjectID)
+	} else {
+		theaterId = update.UpsertedID.(primitive.ObjectID)
+	}
+
+	// sending new media source to websocket
+	err = internal.Client.TheaterService.SendMediaSourceUpdateEvent(insertedID.Hex(), theaterId.Hex())
+	if err != nil {
+		sentry.CaptureException(err)
 	}
 
 	return &proto.Response{
@@ -144,6 +206,52 @@ func (s *Service) AddMediaSource(ctx context.Context, req *proto.MediaSourceAuth
 	}, nil
 }
 
+func (s *Service) GetMediaSource(ctx context.Context, req *proto.MediaSourceAuthRequest) (*proto.TheaterMediaSourcesResponse, error) {
+
+	var (
+		database     = db.Connection
+		mediaSources = make([]*proto.MediaSource, 0)
+	)
+
+	user, err := auth.Authenticate(req.AuthRequest)
+	if err != nil {
+		return &proto.TheaterMediaSourcesResponse{
+			Status:  "failed",
+			Code:    http.StatusUnauthorized,
+			Message: "Unauthorized!",
+		}, nil
+	}
+
+	mediaSourceObjectId, err := primitive.ObjectIDFromHex(req.Media.Id)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, "MediaSourceId is invalid!")
+	}
+
+	var (
+		mediaSource = new(models.MediaSource)
+		filter = bson.M{
+			"user_id": user.ID,
+			"_id": mediaSourceObjectId,
+		}
+	)
+
+	if err := database.Collection("media_sources").FindOne(ctx, filter).Decode(mediaSource); err != nil {
+		return nil, status.Error(codes.NotFound, "Could not find media source!")
+	}
+
+	protoMediaSource, err := helpers.NewMediaSourceProto(mediaSource)
+	if err != nil {
+		return nil, status.Error(codes.Internal, "Could not parse media source!")
+	}
+	mediaSources = append(mediaSources, protoMediaSource)
+
+	return &proto.TheaterMediaSourcesResponse{
+		Status:  "success",
+		Code:    http.StatusOK,
+		Result:  mediaSources,
+	}, nil
+}
+
 func (s *Service) GetMediaSources(ctx context.Context, req *proto.MediaSourceAuthRequest) (*proto.TheaterMediaSourcesResponse, error) {
 
 	var (
@@ -202,8 +310,24 @@ func (s *Service) RemoveMediaSource(ctx context.Context, req *proto.MediaSourceR
 		return nil, status.Error(codes.Internal, "Could not parse MediaSourceId!")
 	}
 
+	var (
+		theater = new(models.Theater)
+		findTheater = bson.M{ "user_id": user.ID }
+	)
+
+	if err := db.Connection.Collection("theaters").FindOne(ctx, findTheater).Decode(theater); err != nil {
+		return nil, status.Error(codes.Internal, "Could not find theater!")
+	}
+
 	if result, err := collection.DeleteOne(ctx, bson.M{ "_id": mediaSourceObjectID, "user_id": user.ID }); err == nil {
 		if result.DeletedCount == 1 {
+
+			// sending new media source to websocket
+			err := internal.Client.TheaterService.SendMediaSourceUpdateEvent(req.MediaSourceId, theater.ID.Hex())
+			if err != nil {
+				sentry.CaptureException(err)
+			}
+
 			return &proto.Response{
 				Status:  "success",
 				Code:    http.StatusOK,

services/theater/service.go

@@ -1,3 +1,7 @@
 package theater
 
-type Service struct {}
+import "go.mongodb.org/mongo-driver/mongo"
+
+type Service struct {
+	db *mongo.Database
+}