Only one streaming key for each course

.github/workflows/codeql-analysis.yml

@@ -44,8 +44,8 @@ jobs:
 
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
-    - name: Autobuild
-      uses: github/codeql-action/autobuild@v2
+    #- name: Autobuild
+    #  uses: github/codeql-action/autobuild@v2
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -54,7 +54,7 @@ jobs:
     #    and modify them (or add more) to build your code if your project
     #    uses a compiled language
 
-    #- run: |
+    - run: make all
     #   make bootstrap
     #   make release
 

.github/workflows/go-test.yml

@@ -8,7 +8,7 @@ jobs:
   test:
     strategy:
       matrix:
-        go-version: [ 1.20.x ]
+        go-version: [ 1.21.x ]
         os: [ ubuntu-latest ]
     runs-on: ${{ matrix.os }}
     steps:

.github/workflows/golangci-lint.yml

@@ -15,7 +15,7 @@ jobs:
     steps:
       - uses: actions/setup-go@v4
         with:
-          go-version: 1.19
+          go-version: 1.21
       - uses: actions/checkout@v3
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3

.golangci.yml

@@ -25,7 +25,7 @@ linters:
   fast: false
 
 run:
-  go: '1.19'
+  go: '1.21'
   timeout: 10m
   skip-dirs:
     - node_modules
@@ -67,7 +67,7 @@ linters-settings:
       - name: modifies-value-receiver
   gofumpt:
     extra-rules: false
-    lang-version: "1.19"
+    lang-version: "1.21"
 
 issues:
   max-issues-per-linter: 0

api/courseimport.go

@@ -84,6 +84,7 @@ func (r lectureHallRoutes) postSchedule(c *gin.Context) {
 			Streams:             nil,
 			Users:               nil,
 			Token:               token,
+			StreamKey:           strings.ReplaceAll(uuid.NewV4().String(), "-", "")[:15],
 		}
 
 		var streams []model.Stream
@@ -98,12 +99,21 @@ func (r lectureHallRoutes) postSchedule(c *gin.Context) {
 			if err == nil {
 				eventID = uint(eventIDInt)
 			}
+
+			// When defined use course wide stream key
+			var streamKey string
+			if course.StreamKey == "" {
+				streamKey = strings.ReplaceAll(uuid.NewV4().String(), "-", "")[:15]
+			} else {
+				streamKey = course.StreamKey
+			}
+
 			streams = append(streams, model.Stream{
 				Start:            event.Start,
 				End:              event.End,
 				RoomName:         event.RoomName,
 				LectureHallID:    lectureHall.ID,
-				StreamKey:        strings.ReplaceAll(uuid.NewV4().String(), "-", "")[:15],
+				StreamKey:        streamKey,
 				TUMOnlineEventID: eventID,
 			})
 		}

api/courses.go

@@ -73,6 +73,7 @@ func configGinCourseRouter(router *gin.Engine, daoWrapper dao.DaoWrapper) {
 			courses.PUT("/updateDescription/:streamID", routes.updateDescription)
 			courses.DELETE("/deleteLectureSeries/:streamID", routes.deleteLectureSeries)
 			courses.POST("/submitCut", routes.submitCut)
+			courses.POST("/regenerateKey", routes.regenerateCourseKey)
 
 			courses.POST("/addUnit", routes.addUnit)
 			courses.POST("/deleteUnit/:unitID", routes.deleteUnit)
@@ -1234,8 +1235,13 @@ func (r coursesRoutes) createLecture(c *gin.Context) {
 	for _, date := range req.DateSeries {
 		endTime := date.Add(time.Minute * time.Duration(req.Duration))
 
-		streamKey := uuid.NewV4().String()
-		streamKey = strings.ReplaceAll(streamKey, "-", "")
+		// When defined use course wide stream key
+		var streamKey string
+		if tumLiveContext.Course.StreamKey == "" {
+			streamKey = strings.ReplaceAll(uuid.NewV4().String(), "-", "")
+		} else {
+			streamKey = tumLiveContext.Course.StreamKey
+		}
 
 		lecture := model.Stream{
 			Name:          req.Title,
@@ -1391,6 +1397,7 @@ func (r coursesRoutes) createCourse(c *gin.Context) {
 		ChatEnabled:         req.EnChat,
 		Visibility:          req.Access,
 		Streams:             []model.Stream{},
+		StreamKey:           strings.ReplaceAll(uuid.NewV4().String(), "-", ""),
 	}
 	if tumLiveContext.User.Role != model.AdminType {
 		course.Admins = []model.User{*tumLiveContext.User}
@@ -1484,8 +1491,31 @@ func (r coursesRoutes) deleteCourse(c *gin.Context) {
 	dao.Cache.Clear()
 }
 
+// regenerateCourseKey updates the stream key of the course and updates the key of all the streams
+func (r coursesRoutes) regenerateCourseKey(c *gin.Context) {
+	ctx := c.MustGet("TUMLiveContext").(tools.TUMLiveContext)
+	course := *ctx.Course
+
+	course.StreamKey = strings.ReplaceAll(uuid.NewV4().String(), "-", "")
+	err := r.DaoWrapper.CoursesDao.UpdateCourse(c, course)
+	if err != nil {
+		c.AbortWithStatusJSON(http.StatusBadRequest, "could not update course")
+		return
+	}
+
+	for _, s := range course.Streams {
+		s.StreamKey = course.StreamKey
+
+		err := r.DaoWrapper.StreamsDao.UpdateStream(s)
+		// Log error but continue on remaining streams
+		if err != nil {
+			log.Error("could not modify stream key ", err)
+		}
+	}
+}
+
 type createCourseRequest struct {
-	Access       string //enrolled, public, hidden or loggedin
+	Access       string //enrolled, public, hidden or logged in
 	CourseID     string
 	EnChat       bool
 	EnDL         bool

api/runner_grpc.go

@@ -0,0 +1,73 @@
+package api
+
+import (
+	"context"
+	"database/sql"
+	"github.com/TUM-Dev/gocast/dao"
+	"github.com/TUM-Dev/gocast/model"
+	log "github.com/sirupsen/logrus"
+	"github.com/tum-dev/gocast/runner/protobuf"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/reflection"
+	"net"
+	"time"
+)
+
+var _ protobuf.FromRunnerServer = (*GrpcRunnerServer)(nil)
+
+type GrpcRunnerServer struct {
+	protobuf.UnimplementedFromRunnerServer
+
+	dao.DaoWrapper
+}
+
+func (g GrpcRunnerServer) Register(ctx context.Context, request *protobuf.RegisterRequest) (*protobuf.RegisterResponse, error) {
+	runner := model.Runner{
+		Hostname: request.Hostname,
+		Port:     int(request.Port),
+		LastSeen: sql.NullTime{Valid: true, Time: time.Now()},
+	}
+	err := g.RunnerDao.Create(ctx, &runner)
+	if err != nil {
+		return nil, err
+	}
+	return &protobuf.RegisterResponse{}, nil
+}
+
+func (g GrpcRunnerServer) Heartbeat(ctx context.Context, request *protobuf.HeartbeatRequest) (*protobuf.HeartbeatResponse, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (g GrpcRunnerServer) RequestSelfStream(ctx context.Context, request *protobuf.SelfStreamRequest) (*protobuf.SelfStreamResponse, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (g GrpcRunnerServer) mustEmbedUnimplementedFromRunnerServer() {
+	//TODO implement me
+	panic("implement me")
+}
+
+func StartGrpcRunnerServer() {
+	lis, err := net.Listen("tcp", ":50056")
+	if err != nil {
+		log.WithError(err).Error("Failed to init grpc server")
+		return
+	}
+	grpcServer := grpc.NewServer(grpc.KeepaliveParams(keepalive.ServerParameters{
+		MaxConnectionIdle:     time.Minute,
+		MaxConnectionAge:      time.Minute * 5,
+		MaxConnectionAgeGrace: time.Second * 5,
+		Time:                  time.Minute * 10,
+		Timeout:               time.Second * 20,
+	}))
+	protobuf.RegisterFromRunnerServer(grpcServer, &GrpcRunnerServer{DaoWrapper: dao.NewDaoWrapper()})
+	reflection.Register(grpcServer)
+	go func() {
+		if err = grpcServer.Serve(lis); err != nil {
+			log.WithError(err).Errorf("Can't serve grpc")
+		}
+	}()
+}

api/stream.go

@@ -44,6 +44,8 @@ func configGinStreamRestRouter(router *gin.Engine, daoWrapper dao.DaoWrapper) {
 			streamById.GET("/subtitles/:lang", routes.getSubtitles)
 
 			streamById.GET("/playlist", routes.getStreamPlaylist)
+			streamById.POST("/regenerateKey", routes.regenerateKey)
+			streamById.POST("/restoreKey", routes.restoreKey)
 
 			thumbs := streamById.Group("/thumbs")
 			{
@@ -839,6 +841,33 @@ func (r streamRoutes) updateStreamVisibility(c *gin.Context) {
 	}
 }
 
+// regenerateKey regenerates the key for a stream.
+func (r streamRoutes) regenerateKey(c *gin.Context) {
+	ctx := c.MustGet("TUMLiveContext").(tools.TUMLiveContext)
+	stream := *ctx.Stream
+
+	stream.StreamKey = strings.ReplaceAll(uuid.NewV4().String(), "-", "")
+	err := r.DaoWrapper.StreamsDao.UpdateStream(stream)
+	if err != nil {
+		c.AbortWithStatusJSON(http.StatusBadRequest, "could not update stream")
+		return
+	}
+}
+
+// restoreKey restores the key for a stream to the course key
+func (r streamRoutes) restoreKey(c *gin.Context) {
+	ctx := c.MustGet("TUMLiveContext").(tools.TUMLiveContext)
+	stream := *ctx.Stream
+	course := *ctx.Course
+
+	stream.StreamKey = course.StreamKey
+	err := r.DaoWrapper.StreamsDao.UpdateStream(stream)
+	if err != nil {
+		c.AbortWithStatusJSON(http.StatusBadRequest, "could not update stream")
+		return
+	}
+}
+
 func (r streamRoutes) updateChatEnabled(c *gin.Context) {
 	stream, err := r.StreamsDao.GetStreamByID(context.Background(), c.Param("streamID"))
 	if err != nil {

api/worker_grpc.go

@@ -19,12 +19,12 @@ import (
 	"time"
 
 	go_anel_pwrctrl "github.com/RBG-TUM/go-anel-pwrctrl"
-	"github.com/getsentry/sentry-go"
 	"github.com/TUM-Dev/gocast/dao"
 	"github.com/TUM-Dev/gocast/model"
 	"github.com/TUM-Dev/gocast/tools"
 	"github.com/TUM-Dev/gocast/tools/camera"
 	"github.com/TUM-Dev/gocast/worker/pb"
+	"github.com/getsentry/sentry-go"
 	uuid "github.com/satori/go.uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
@@ -128,22 +128,18 @@ func (s server) SendSelfStreamRequest(ctx context.Context, request *pb.SelfStrea
 	if request.StreamKey == "" {
 		return nil, errors.New("stream key empty")
 	}
-	stream, err := s.StreamsDao.GetStreamByKey(ctx, request.StreamKey)
+	stream, err := s.StreamsDao.GetStreamByKeyAndTime(ctx, request.StreamKey, time.Now())
 	if err != nil {
 		return nil, err
 	}
 	course, err := s.DaoWrapper.CoursesDao.GetCourseById(ctx, stream.CourseID)
 	if err != nil {
 		return nil, err
 	}
-	if request.CourseSlug != fmt.Sprintf("%s-%d", course.Slug, stream.ID) {
-		return nil, fmt.Errorf("bad stream name, should: %s, is: %s", fmt.Sprintf("%s-%d", course.Slug, stream.ID), request.CourseSlug)
-	}
-	// reject streams that are more than 30 minutes in the future or more than 30 minutes past
-	if !(time.Now().After(stream.Start.Add(time.Minute*-30)) && time.Now().Before(stream.End.Add(time.Minute*30))) {
-		log.WithFields(log.Fields{"streamId": stream.ID}).Warn("Stream rejected, time out of bounds")
-		return nil, errors.New("stream rejected")
+	if request.CourseSlug != fmt.Sprintf("%s", course.Slug) {
+		return nil, fmt.Errorf("bad stream name, should: %s, is: %s", fmt.Sprintf("%s", course.Slug), request.CourseSlug)
 	}
+
 	ingestServer, err := s.DaoWrapper.IngestServerDao.GetBestIngestServer()
 	if err != nil {
 		return nil, err

cmd/tumlive/tumlive.go

@@ -2,17 +2,17 @@ package main
 
 import (
 	"fmt"
-	"github.com/dgraph-io/ristretto"
-	"github.com/getsentry/sentry-go"
-	sentrygin "github.com/getsentry/sentry-go/gin"
-	"github.com/gin-contrib/gzip"
-	"github.com/gin-gonic/gin"
 	"github.com/TUM-Dev/gocast/api"
 	"github.com/TUM-Dev/gocast/dao"
 	"github.com/TUM-Dev/gocast/model"
 	"github.com/TUM-Dev/gocast/tools"
 	"github.com/TUM-Dev/gocast/tools/tum"
 	"github.com/TUM-Dev/gocast/web"
+	"github.com/dgraph-io/ristretto"
+	"github.com/getsentry/sentry-go"
+	sentrygin "github.com/getsentry/sentry-go/gin"
+	"github.com/gin-contrib/gzip"
+	"github.com/gin-gonic/gin"
 	"github.com/pkg/profile"
 	log "github.com/sirupsen/logrus"
 	"gorm.io/driver/mysql"
@@ -32,6 +32,7 @@ type initializer func()
 var initializers = []initializer{
 	tools.LoadConfig,
 	api.ServeWorkerGRPC,
+	api.StartGrpcRunnerServer,
 	tools.InitBranding,
 }
 
@@ -168,6 +169,7 @@ func main() {
 		&model.Subtitles{},
 		&model.TranscodingFailure{},
 		&model.Email{},
+		&model.Runner{},
 	)
 	if err != nil {
 		sentry.CaptureException(err)

dao/dao_base.go

@@ -36,6 +36,7 @@ type DaoWrapper struct {
 	SubtitlesDao
 	TranscodingFailureDao
 	EmailDao
+	RunnerDao RunnerDao
 }
 
 func NewDaoWrapper() DaoWrapper {
@@ -63,5 +64,6 @@ func NewDaoWrapper() DaoWrapper {
 		SubtitlesDao:          NewSubtitlesDao(),
 		TranscodingFailureDao: NewTranscodingFailureDao(),
 		EmailDao:              NewEmailDao(),
+		RunnerDao:             NewRunnerDao(),
 	}
 }