added 30 min buffer to finding stream by time

api/worker_grpc.go

@@ -132,21 +132,14 @@ func (s server) SendSelfStreamRequest(ctx context.Context, request *pb.SelfStrea
 	if err != nil {
 		return nil, err
 	}
-	log.Printf("Got stream: %+v\n", stream)
 	course, err := s.DaoWrapper.CoursesDao.GetCourseById(ctx, stream.CourseID)
 	if err != nil {
 		return nil, err
 	}
 	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)
 	}
-	/*
-		// 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")
-		}
-	*/
+
 	ingestServer, err := s.DaoWrapper.IngestServerDao.GetBestIngestServer()
 	if err != nil {
 		return nil, err

dao/streams.go

@@ -146,13 +146,13 @@ func (d streamsDao) GetStreamByKeyAndTime(ctx context.Context, key string, t tim
 	if result.Error != nil {
 		return model.Stream{}, err
 	}
-	// find stream that is at the latest in 30 minutes
+	// find stream that encompasses the given time with 30 minutes of padding before and after
 	for _, s := range streams {
-		if t.Add(time.Minute*-30).After(s.Start) && t.Before(s.End) {
+		if t.After(stream.Start.Add(time.Minute*-30)) && t.Before(stream.End.Add(time.Minute*30)) {
 			return s, err
 		}
 	}
-	return model.Stream{}, gorm.ErrRecordNotFound
+	return model.Stream{}, fmt.Errorf("no stream at %s", t.String())
 }
 
 func (d streamsDao) GetUnitByID(id string) (model.StreamUnit, error) {