Merge branch 'replays-clips-recordings' into gek/initial-replay-functionality

* chore(deps): update dependency typescript to v5

* fix: resolve type mismatch

* Commit updated API documentation

---------

Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
Co-authored-by: Gabe Kangas <gabek@real-ity.com>
Co-authored-by: Owncast <owncast@owncast.online>

config/config.go

@@ -43,6 +43,9 @@ var EnableAutoUpdate = false
 // A temporary stream key that can be set via the command line.
 var TemporaryStreamKey = ""
 
+// EnableReplayFeatures will enable replay features.
+var EnableReplayFeatures = true
+
 // GetCommit will return an identifier used for identifying the point in time this build took place.
 func GetCommit() string {
 	if GitCommit == "" {

controllers/admin/config.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/owncast/owncast/activitypub/outbox"
 	"github.com/owncast/owncast/controllers"
+	"github.com/owncast/owncast/core"
 	"github.com/owncast/owncast/core/chat"
 	"github.com/owncast/owncast/core/data"
 	"github.com/owncast/owncast/core/user"
@@ -76,8 +77,10 @@ func SetStreamTitle(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	if value != "" {
+		streamID := core.GetCurrentBroadcast().StreamID
+
 		sendSystemChatAction(fmt.Sprintf("Stream title changed to **%s**", value), true)
-		go webhooks.SendStreamStatusEvent(models.StreamTitleUpdated)
+		go webhooks.SendStreamStatusEvent(models.StreamTitleUpdated, streamID)
 	}
 	controllers.WriteSimpleResponse(w, true, "changed")
 }

controllers/clips.go

@@ -0,0 +1,167 @@
+package controllers
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/owncast/owncast/config"
+	"github.com/owncast/owncast/replays"
+	log "github.com/sirupsen/logrus"
+)
+
+// GetAllClips will return all clips that have been previously created.
+func GetAllClips(w http.ResponseWriter, r *http.Request) {
+	if !config.EnableReplayFeatures {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	clips, err := replays.GetAllClips()
+	if err != nil {
+		log.Errorln(err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	WriteResponse(w, clips)
+}
+
+// AddClip will create a new clip for a given stream and time window.
+func AddClip(w http.ResponseWriter, r *http.Request) {
+	if !config.EnableReplayFeatures {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	type addClipRequest struct {
+		StreamId                 string  `json:"streamId"`
+		ClipTitle                string  `json:"clipTitle"`
+		RelativeStartTimeSeconds float32 `json:"relativeStartTimeSeconds"`
+		RelativeEndTimeSeconds   float32 `json:"relativeEndTimeSeconds"`
+	}
+
+	if r.Method != http.MethodPost {
+		BadRequestHandler(w, nil)
+		return
+	}
+
+	decoder := json.NewDecoder(r.Body)
+	var request addClipRequest
+
+	if request.RelativeEndTimeSeconds < request.RelativeStartTimeSeconds {
+		BadRequestHandler(w, errors.New("end time must be after start time"))
+		return
+	}
+
+	if err := decoder.Decode(&request); err != nil {
+		log.Errorln(err)
+		WriteSimpleResponse(w, false, "unable to create clip")
+		return
+	}
+
+	streamId := request.StreamId
+	clipTitle := request.ClipTitle
+	startTime := request.RelativeStartTimeSeconds
+	endTime := request.RelativeEndTimeSeconds
+
+	// Some validation
+	playlistGenerator := replays.NewPlaylistGenerator()
+
+	stream, err := playlistGenerator.GetStream(streamId)
+	if err != nil {
+		BadRequestHandler(w, errors.New("stream not found"))
+		return
+	}
+
+	if stream.StartTime.IsZero() {
+		BadRequestHandler(w, errors.New("stream start time not found"))
+		return
+	}
+
+	// Make sure the proposed clip start time and end time are within
+	// the start and end time of the stream.
+	finalSegment, err := replays.GetFinalSegmentForStream(streamId)
+	if err != nil {
+		InternalErrorHandler(w, err)
+		return
+	}
+
+	if finalSegment.RelativeTimestamp < startTime {
+		BadRequestHandler(w, errors.New("start time is after the known end of the stream"))
+		return
+	}
+
+	clipId, duration, err := replays.AddClipForStream(streamId, clipTitle, "", startTime, endTime)
+	if err != nil {
+		log.Errorln(err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	WriteSimpleResponse(w, true, "clip "+clipId+" created with duration of "+fmt.Sprint(duration)+" seconds")
+}
+
+// GetClip will return playable content for a given clip Id.
+func GetClip(w http.ResponseWriter, r *http.Request) {
+	if !config.EnableReplayFeatures {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	pathComponents := strings.Split(r.URL.Path, "/")
+	if len(pathComponents) == 3 {
+		// Return the master playlist for the requested stream
+		clipId := pathComponents[2]
+		getClipMasterPlaylist(clipId, w)
+		return
+	} else if len(pathComponents) == 4 {
+		// Return the media playlist for the requested stream and output config
+		clipId := pathComponents[2]
+		outputConfigId := pathComponents[3]
+		getClipMediaPlaylist(clipId, outputConfigId, w)
+		return
+	}
+
+	BadRequestHandler(w, nil)
+}
+
+// getReplayMasterPlaylist will return a complete replay of a stream
+// as a HLS playlist.
+func getClipMasterPlaylist(clipId string, w http.ResponseWriter) {
+	playlistGenerator := replays.NewPlaylistGenerator()
+	playlist, err := playlistGenerator.GenerateMasterPlaylistForClip(clipId)
+	if err != nil {
+		log.Println(err)
+	}
+
+	if playlist == nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	w.Header().Add("Content-Type", "application/x-mpegURL")
+	if _, err := w.Write(playlist.Encode().Bytes()); err != nil {
+		log.Errorln(err)
+		return
+	}
+}
+
+// getClipMediaPlaylist will return media playlist for a given clip
+// and stream output configuration.
+func getClipMediaPlaylist(clipId, outputConfigId string, w http.ResponseWriter) {
+	playlistGenerator := replays.NewPlaylistGenerator()
+	playlist, err := playlistGenerator.GenerateMediaPlaylistForClipAndConfiguration(clipId, outputConfigId)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Add("Content-Type", "application/x-mpegURL")
+	if _, err := w.Write(playlist.Encode().Bytes()); err != nil {
+		log.Errorln(err)
+		return
+	}
+}

controllers/replays.go

@@ -0,0 +1,84 @@
+package controllers
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/owncast/owncast/config"
+	"github.com/owncast/owncast/replays"
+	log "github.com/sirupsen/logrus"
+)
+
+// GetReplays will return a list of all available replays.
+func GetReplays(w http.ResponseWriter, r *http.Request) {
+	if !config.EnableReplayFeatures {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	streams, err := replays.GetStreams()
+	if err != nil {
+		log.Errorln(err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	WriteResponse(w, streams)
+}
+
+// GetReplay will return a playable content for a given stream Id.
+func GetReplay(w http.ResponseWriter, r *http.Request) {
+	pathComponents := strings.Split(r.URL.Path, "/")
+	if len(pathComponents) == 3 {
+		// Return the master playlist for the requested stream
+		streamId := pathComponents[2]
+		getReplayMasterPlaylist(streamId, w)
+		return
+	} else if len(pathComponents) == 4 {
+		// Return the media playlist for the requested stream and output config
+		streamId := pathComponents[2]
+		outputConfigId := pathComponents[3]
+		getReplayMediaPlaylist(streamId, outputConfigId, w)
+		return
+	}
+
+	BadRequestHandler(w, nil)
+}
+
+// getReplayMasterPlaylist will return a complete replay of a stream as a HLS playlist.
+// /api/replay/{streamId}.
+func getReplayMasterPlaylist(streamId string, w http.ResponseWriter) {
+	playlistGenerator := replays.NewPlaylistGenerator()
+	playlist, err := playlistGenerator.GenerateMasterPlaylistForStream(streamId)
+	if err != nil {
+		log.Println(err)
+	}
+
+	if playlist == nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	w.Header().Add("Content-Type", "application/x-mpegURL")
+	if _, err := w.Write(playlist.Encode().Bytes()); err != nil {
+		log.Errorln(err)
+		return
+	}
+}
+
+// getReplayMediaPlaylist will return a media playlist for a given stream.
+// /api/replay/{streamId}/{outputConfigId}.
+func getReplayMediaPlaylist(streamId, outputConfigId string, w http.ResponseWriter) {
+	playlistGenerator := replays.NewPlaylistGenerator()
+	playlist, err := playlistGenerator.GenerateMediaPlaylistForStreamAndConfiguration(streamId, outputConfigId)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Add("Content-Type", "application/x-mpegURL")
+	if _, err := w.Write(playlist.Encode().Bytes()); err != nil {
+		log.Errorln(err)
+		return
+	}
+}

core/core.go

@@ -17,6 +17,7 @@ import (
 	"github.com/owncast/owncast/core/webhooks"
 	"github.com/owncast/owncast/models"
 	"github.com/owncast/owncast/notifications"
+	"github.com/owncast/owncast/replays"
 	"github.com/owncast/owncast/utils"
 	"github.com/owncast/owncast/yp"
 )
@@ -58,6 +59,7 @@ func Start() error {
 
 	user.SetupUsers()
 	auth.Setup(data.GetDatastore())
+	replays.Setup()
 
 	fileWriter.SetupFileWriterReceiverService(&handler)
 
@@ -99,8 +101,12 @@ func createInitialOfflineState() error {
 func transitionToOfflineVideoStreamContent() {
 	log.Traceln("Firing transcoder with offline stream state")
 
-	_transcoder := transcoder.NewTranscoder()
+	streamId := "offline"
-	_transcoder.SetIdentifier("offline")
+	_storage.SetStreamId(streamId)
+	fileWriter.SetStreamID(streamId)
+	handler.SetStreamId(streamId)
+
+	_transcoder := transcoder.NewTranscoder(streamId)
 	_transcoder.SetLatencyLevel(models.GetLatencyLevel(4))
 	_transcoder.SetIsEvent(true)
 
@@ -127,7 +133,7 @@ func resetDirectories() {
 	log.Trace("Resetting file directories to a clean slate.")
 
 	// Wipe hls data directory
-	utils.CleanupDirectory(config.HLSStoragePath)
+	utils.CleanupDirectory(config.HLSStoragePath, config.EnableReplayFeatures)
 
 	// Remove the previous thumbnail
 	logo := data.GetLogoPath()

core/data/data.go

@@ -76,6 +76,7 @@ func SetupPersistence(file string) error {
 	createWebhooksTable()
 	createUsersTable(db)
 	createAccessTokenTable(db)
+	createRecordingTables(db)
 
 	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS config (
 		"key" string NOT NULL PRIMARY KEY,

core/data/replays.go

@@ -0,0 +1,64 @@
+package data
+
+import (
+	"database/sql"
+)
+
+func createRecordingTables(db *sql.DB) {
+	createSegmentsTableSQL := `CREATE TABLE IF NOT EXISTS video_segments (
+		"id" string NOT NULL,
+		"stream_id" string NOT NULL,
+		"output_configuration_id" string NOT NULL,
+		"path" TEXT NOT NULL,
+		"relative_timestamp" REAL NOT NULL,
+		"timestamp" DATETIME,
+		PRIMARY KEY (id)
+	);CREATE INDEX video_segments_stream_id ON video_segments (stream_id);CREATE INDEX video_segments_stream_id_timestamp ON video_segments (stream_id,timestamp);`
+
+	createVideoOutputConfigsTableSQL := `CREATE TABLE IF NOT EXISTS video_segment_output_configuration (
+		"id" string NOT NULL,
+		"variant_id" string NOT NULL,
+		"name" string NOT NULL,
+		"stream_id" string NOT NULL,
+		"segment_duration" INTEGER NOT NULL,
+		"bitrate" INTEGER NOT NULL,
+		"framerate" INTEGER NOT NULL,
+		"resolution_width" INTEGER,
+		"resolution_height" INTEGER,
+		"timestamp" DATETIME,
+		PRIMARY KEY (id)
+	);CREATE INDEX video_segment_output_configuration_stream_id ON video_segment_output_configuration (stream_id);`
+
+	createVideoStreamsTableSQL := `CREATE TABLE IF NOT EXISTS streams (
+		"id" string NOT NULL,
+		"stream_title" TEXT,
+		"start_time" DATETIME,
+		"end_time" DATETIME,
+		PRIMARY KEY (id)
+	);
+	CREATE INDEX streams_id ON streams (id);
+	CREATE INDEX streams_start_time ON streams (start_time);
+	CREATE INDEX streams_start_end_time ON streams (start_time,end_time);
+	`
+
+	createClipsTableSQL := `CREATE TABLE IF NOT EXISTS replay_clips (
+		"id" string NOT NULL,
+		"stream_id" string NOT NULL,
+		"clipped_by" string,
+		"clip_title" TEXT,
+		"relative_start_time" REAL,
+		"relative_end_time" REAL,
+		"timestamp" DATETIME,
+		PRIMARY KEY (id),
+		FOREIGN KEY(stream_id) REFERENCES streams(id)
+	);
+	CREATE INDEX clip_id ON replay_clips (id);
+	CREATE INDEX clip_stream_id ON replay_clips (stream_id);
+	CREATE INDEX clip_start_end_time ON replay_clips (start_time,end_time);
+	`
+
+	MustExec(createSegmentsTableSQL, db)
+	MustExec(createVideoOutputConfigsTableSQL, db)
+	MustExec(createVideoStreamsTableSQL, db)
+	MustExec(createClipsTableSQL, db)
+}

core/offlineState.go

@@ -48,15 +48,16 @@ func appendOfflineToVariantPlaylist(index int, playlistFilePath string) {
 	}
 }
 
-func makeVariantIndexOffline(index int, offlineFilePath string, offlineFilename string) {
+func makeVariantIndexOffline(streamId string, index int, offlineFilePath string, offlineFilename string) {
-	playlistFilePath := fmt.Sprintf(filepath.Join(config.HLSStoragePath, "%d/stream.m3u8"), index)
+	playlistFilePath := fmt.Sprintf(filepath.Join(config.HLSStoragePath, streamId, "%d/stream.m3u8"), index)
-	segmentFilePath := fmt.Sprintf(filepath.Join(config.HLSStoragePath, "%d/%s"), index, offlineFilename)
+	// segmentFilePath := fmt.Sprintf(filepath.Join(config.HLSStoragePath, streamId, "%d/%s"), index, offlineFilename)
+	segmentFileDestinationPath := fmt.Sprintf(filepath.Join("hls", streamId, "%d/%s"), index, offlineFilename)
 
-	if err := utils.Copy(offlineFilePath, segmentFilePath); err != nil {
+	if err := utils.Copy(offlineFilePath, offlineFilePath); err != nil {
 		log.Warnln(err)
 	}
 
-	if _, err := _storage.Save(segmentFilePath, 0); err != nil {
+	if _, err := _storage.Save(offlineFilePath, segmentFileDestinationPath, 0); err != nil {
 		log.Warnln(err)
 	}
 
@@ -65,7 +66,7 @@ func makeVariantIndexOffline(index int, offlineFilePath string, offlineFilename
 	} else {
 		createEmptyOfflinePlaylist(playlistFilePath, offlineFilename)
 	}
-	if _, err := _storage.Save(playlistFilePath, 0); err != nil {
+	if _, err := _storage.Save(playlistFilePath, playlistFilePath, 0); err != nil {
 		log.Warnln(err)
 	}
 }

core/storageproviders/local.go

@@ -10,10 +10,12 @@ import (
 
 	"github.com/owncast/owncast/config"
 	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/utils"
 )
 
 // LocalStorage represents an instance of the local storage provider for HLS video.
 type LocalStorage struct {
+	streamID string
 	host string
 }
 
@@ -22,6 +24,11 @@ func NewLocalStorage() *LocalStorage {
 	return &LocalStorage{}
 }
 
+// SetStreamId sets the stream id for this storage provider.
+func (s *LocalStorage) SetStreamId(streamID string) {
+	s.streamID = streamID
+}
+
 // Setup configures this storage provider.
 func (s *LocalStorage) Setup() error {
 	s.host = data.GetVideoServingEndpoint()
@@ -29,15 +36,27 @@ func (s *LocalStorage) Setup() error {
 }
 
 // SegmentWritten is called when a single segment of video is written.
-func (s *LocalStorage) SegmentWritten(localFilePath string) {
+func (s *LocalStorage) SegmentWritten(localFilePath string) (string, int, error) {
-	if _, err := s.Save(localFilePath, 0); err != nil {
+	if s.streamID == "" {
+		log.Fatalln("stream id must be set when handling video segments")
+	}
+
+	destinationPath, err := s.Save(localFilePath, localFilePath, 0)
+	if err != nil {
 		log.Warnln(err)
+		return "", 0, err
 	}
+
+	return destinationPath, 0, nil
 }
 
 // VariantPlaylistWritten is called when a variant hls playlist is written.
 func (s *LocalStorage) VariantPlaylistWritten(localFilePath string) {
-	if _, err := s.Save(localFilePath, 0); err != nil {
+	if s.streamID == "" {
+		log.Fatalln("stream id must be set when handling video playlists")
+	}
+
+	if _, err := s.Save(localFilePath, localFilePath, 0); err != nil {
 		log.Errorln(err)
 		return
 	}
@@ -45,28 +64,38 @@ func (s *LocalStorage) VariantPlaylistWritten(localFilePath string) {
 
 // MasterPlaylistWritten is called when the master hls playlist is written.
 func (s *LocalStorage) MasterPlaylistWritten(localFilePath string) {
-	// If we're using a remote serving endpoint, we need to rewrite the master playlist
+	if s.streamID == "" {
-	if s.host != "" {
+		log.Fatalln("stream id must be set when handling video playlists")
-		if err := rewritePlaylistLocations(localFilePath, s.host, ""); err != nil {
+	}
-			log.Warnln(err)
+
-		}
+	masterPlaylistDestinationLocation := filepath.Join(config.HLSStoragePath, "/stream.m3u8")
-	} else {
+	if err := rewriteLocalPlaylist(localFilePath, s.streamID, masterPlaylistDestinationLocation); err != nil {
-		if _, err := s.Save(localFilePath, 0); err != nil {
+		log.Errorln(err)
-			log.Warnln(err)
+		return
-		}
 	}
 }
 
 // Save will save a local filepath using the storage provider.
-func (s *LocalStorage) Save(filePath string, retryCount int) (string, error) {
+func (s *LocalStorage) Save(filePath, destinationPath string, retryCount int) (string, error) {
-	return filePath, nil
+	if filePath != destinationPath {
+		if err := utils.Move(filePath, destinationPath); err != nil {
+			return "", errors.Wrap(err, "unable to move file")
+		}
+	}
+
+	return destinationPath, nil
 }
 
 func (s *LocalStorage) Cleanup() error {
+	// If we're recording, don't perform the cleanup.
+	if config.EnableReplayFeatures {
+		return nil
+	}
+
 	// Determine how many files we should keep on disk
 	maxNumber := data.GetStreamLatencyLevel().SegmentCount
 	buffer := 10
-	baseDirectory := config.HLSStoragePath
+	baseDirectory := filepath.Join(config.HLSStoragePath, s.streamID)
 
 	files, err := getAllFilesRecursive(baseDirectory)
 	if err != nil {
@@ -94,6 +123,10 @@ func (s *LocalStorage) Cleanup() error {
 	return nil
 }
 
+func (s *LocalStorage) GetRemoteDestinationPathFromLocalFilePath(localFilePath string) string {
+	return localFilePath
+}
+
 func getAllFilesRecursive(baseDirectory string) (map[string][]os.FileInfo, error) {
 	files := make(map[string][]os.FileInfo)
 

core/storageproviders/rewriteLocalPlaylist.go

@@ -25,14 +25,7 @@ func rewritePlaylistLocations(localFilePath, remoteServingEndpoint, pathPrefix s
 	}
 
 	for _, item := range p.Variants {
-		// Determine the final path to this playlist.
+		item.URI = filepath.Join(remoteServingEndpoint, pathPrefix, item.URI)
-		var finalPath string
-		if pathPrefix != "" {
-			finalPath = filepath.Join(pathPrefix, "/hls")
-		} else {
-			finalPath = "/hls"
-		}
-		item.URI = remoteServingEndpoint + filepath.Join(finalPath, item.URI)
 	}
 
 	publicPath := filepath.Join(config.HLSStoragePath, filepath.Base(localFilePath))
@@ -41,3 +34,29 @@ func rewritePlaylistLocations(localFilePath, remoteServingEndpoint, pathPrefix s
 
 	return playlist.WritePlaylist(newPlaylist, publicPath)
 }
+
+// rewriteLocalPlaylist will take a local master playlist and rewrite it to
+// refer to the path that includes the stream ID.
+func rewriteLocalPlaylist(localFilePath, streamID, destinationPath string) error {
+	f, err := os.Open(localFilePath) // nolint
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	p := m3u8.NewMasterPlaylist()
+	if err := p.DecodeFrom(bufio.NewReader(f), false); err != nil {
+		log.Warnln(err)
+	}
+
+	if streamID == "" {
+		log.Fatalln("stream id must be set when rewriting playlist contents")
+	}
+
+	for _, item := range p.Variants {
+		item.URI = filepath.Join("/hls", streamID, item.URI)
+	}
+
+	newPlaylist := p.String()
+
+	return playlist.WritePlaylist(newPlaylist, destinationPath)
+}

core/storageproviders/s3Storage.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/owncast/owncast/config"
 	"github.com/owncast/owncast/core/data"
 	"github.com/owncast/owncast/utils"
 	"github.com/pkg/errors"
@@ -20,12 +21,11 @@ import (
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
-
-	"github.com/owncast/owncast/config"
 )
 
 // S3Storage is the s3 implementation of a storage provider.
 type S3Storage struct {
+	streamId string
 	sess     *session.Session
 	s3Client *s3.S3
 
@@ -56,6 +56,11 @@ func NewS3Storage() *S3Storage {
 	}
 }
 
+// SetStreamID sets the stream id for this storage provider.
+func (s *S3Storage) SetStreamId(streamId string) {
+	s.streamId = streamId
+}
+
 // Setup sets up the s3 storage for saving the video to s3.
 func (s *S3Storage) Setup() error {
 	log.Trace("Setting up S3 for external storage of video...")
@@ -88,16 +93,19 @@ func (s *S3Storage) Setup() error {
 }
 
 // SegmentWritten is called when a single segment of video is written.
-func (s *S3Storage) SegmentWritten(localFilePath string) {
+func (s *S3Storage) SegmentWritten(localFilePath string) (string, int, error) {
 	index := utils.GetIndexFromFilePath(localFilePath)
 	performanceMonitorKey := "s3upload-" + index
 	utils.StartPerformanceMonitor(performanceMonitorKey)
 
 	// Upload the segment
-	if _, err := s.Save(localFilePath, 0); err != nil {
+	remoteDestinationPath := s.GetRemoteDestinationPathFromLocalFilePath(localFilePath)
+	remotePath, err := s.Save(localFilePath, remoteDestinationPath, 0)
+	if err != nil {
 		log.Errorln(err)
-		return
+		return "", 0, err
 	}
+
 	averagePerformance := utils.GetAveragePerformance(performanceMonitorKey)
 
 	// Warn the user about long-running save operations
@@ -111,14 +119,17 @@ func (s *S3Storage) SegmentWritten(localFilePath string) {
 	// so the segments and the HLS playlist referencing
 	// them are in sync.
 	playlistPath := filepath.Join(filepath.Dir(localFilePath), "stream.m3u8")
+	playlistRemoteDestinationPath := s.GetRemoteDestinationPathFromLocalFilePath(playlistPath)
 
-	if _, err := s.Save(playlistPath, 0); err != nil {
+	if _, err := s.Save(playlistPath, playlistRemoteDestinationPath, 0); err != nil {
 		s.queuedPlaylistUpdates[playlistPath] = playlistPath
 		if pErr, ok := err.(*os.PathError); ok {
 			log.Debugln(pErr.Path, "does not yet exist locally when trying to upload to S3 storage.")
-			return
+			return remotePath, 0, pErr.Err
 		}
 	}
+
+	return remotePath, 0, nil
 }
 
 // VariantPlaylistWritten is called when a variant hls playlist is written.
@@ -127,7 +138,8 @@ func (s *S3Storage) VariantPlaylistWritten(localFilePath string) {
 	// to make sure we're not referring to files in a playlist that don't
 	// yet exist.  See SegmentWritten.
 	if _, ok := s.queuedPlaylistUpdates[localFilePath]; ok {
-		if _, err := s.Save(localFilePath, 0); err != nil {
+		remoteDestinationPath := s.GetRemoteDestinationPathFromLocalFilePath(localFilePath)
+		if _, err := s.Save(localFilePath, remoteDestinationPath, 0); err != nil {
 			log.Errorln(err)
 			s.queuedPlaylistUpdates[localFilePath] = localFilePath
 		}
@@ -138,21 +150,25 @@ func (s *S3Storage) VariantPlaylistWritten(localFilePath string) {
 // MasterPlaylistWritten is called when the master hls playlist is written.
 func (s *S3Storage) MasterPlaylistWritten(localFilePath string) {
 	// Rewrite the playlist to use absolute remote S3 URLs
-	if err := rewritePlaylistLocations(localFilePath, s.host, s.s3PathPrefix); err != nil {
+	pathPrefix := filepath.Join("hls", s.streamId)
+	if s.s3PathPrefix != "" {
+		pathPrefix = filepath.Join(s.s3PathPrefix, pathPrefix)
+	}
+	if err := rewriteRemotePlaylist(localFilePath, s.host, pathPrefix); err != nil {
 		log.Warnln(err)
 	}
 }
 
 // Save saves the file to the s3 bucket.
-func (s *S3Storage) Save(filePath string, retryCount int) (string, error) {
+func (s *S3Storage) Save(localFilePath, remoteDestinationPath string, retryCount int) (string, error) {
-	file, err := os.Open(filePath) // nolint
+	file, err := os.Open(localFilePath) // nolint
 	if err != nil {
 		return "", err
 	}
 	defer file.Close()
 
 	// Convert the local path to the variant/file path by stripping the local storage location.
-	normalizedPath := strings.TrimPrefix(filePath, config.HLSStoragePath)
+	normalizedPath := strings.TrimPrefix(localFilePath, config.HLSStoragePath)
 	// Build the remote path by adding the "hls" path prefix.
 	remotePath := strings.Join([]string{"hls", normalizedPath}, "")
 
@@ -162,7 +178,7 @@ func (s *S3Storage) Save(filePath string, retryCount int) (string, error) {
 		remotePath = strings.Join([]string{prefix, remotePath}, "/")
 	}
 
-	maxAgeSeconds := utils.GetCacheDurationSecondsForPath(filePath)
+	maxAgeSeconds := utils.GetCacheDurationSecondsForPath(localFilePath)
 	cacheControlHeader := fmt.Sprintf("max-age=%d", maxAgeSeconds)
 
 	uploadInput := &s3manager.UploadInput{
@@ -172,7 +188,7 @@ func (s *S3Storage) Save(filePath string, retryCount int) (string, error) {
 		CacheControl: &cacheControlHeader,
 	}
 
-	if path.Ext(filePath) == ".m3u8" {
+	if path.Ext(localFilePath) == ".m3u8" {
 		noCacheHeader := "no-cache, no-store, must-revalidate"
 		contentType := "application/x-mpegURL"
 
@@ -192,22 +208,27 @@ func (s *S3Storage) Save(filePath string, retryCount int) (string, error) {
 		log.Traceln("error uploading segment", err.Error())
 		if retryCount < 4 {
 			log.Traceln("Retrying...")
-			return s.Save(filePath, retryCount+1)
+			return s.Save(localFilePath, remoteDestinationPath, retryCount+1)
 		}
 
 		// Upload failure. Remove the local file.
-		s.removeLocalFile(filePath)
+		s.removeLocalFile(localFilePath)
 
-		return "", fmt.Errorf("Giving up uploading %s to object storage %s", filePath, s.s3Endpoint)
+		return "", fmt.Errorf("Giving up uploading %s to object storage %s", localFilePath, s.s3Endpoint)
 	}
 
 	// Upload success. Remove the local file.
-	s.removeLocalFile(filePath)
+	s.removeLocalFile(localFilePath)
 
 	return response.Location, nil
 }
 
 func (s *S3Storage) Cleanup() error {
+	// If we're recording, don't perform the cleanup.
+	if config.EnableReplayFeatures {
+		return nil
+	}
+
 	// Determine how many files we should keep on S3 storage
 	maxNumber := data.GetStreamLatencyLevel().SegmentCount
 	buffer := 20
@@ -273,7 +294,7 @@ func (s *S3Storage) removeLocalFile(filePath string) {
 	cleanFilepath := filepath.Clean(filePath)
 
 	if err := os.Remove(cleanFilepath); err != nil {
-		log.Errorln(err)
+		log.Debugln(err)
 	}
 }
 
@@ -331,6 +352,16 @@ func (s *S3Storage) retrieveAllVideoSegments() ([]s3object, error) {
 	return allObjects, nil
 }
 
+func (s *S3Storage) GetRemoteDestinationPathFromLocalFilePath(localFilePath string) string {
+	// Convert the local path to the variant/file path by stripping the local storage location.
+	normalizedPath := strings.TrimPrefix(localFilePath, config.HLSStoragePath)
+
+	// Build the remote path by adding the "hls" path prefix.
+	remoteDestionationPath := strings.Join([]string{"hls", normalizedPath}, "")
+
+	return remoteDestionationPath
+}
+
 type s3object struct {
 	lastModified time.Time
 	key          string

core/streamState.go

@@ -3,9 +3,11 @@ package core
 import (
 	"context"
 	"io"
+	"path/filepath"
 	"time"
 
 	log "github.com/sirupsen/logrus"
+	"github.com/teris-io/shortid"
 
 	"github.com/owncast/owncast/activitypub"
 	"github.com/owncast/owncast/config"
@@ -39,36 +41,34 @@ func setStreamAsConnected(rtmpOut *io.PipeReader) {
 	_stats.LastConnectTime = &now
 	_stats.SessionMaxViewerCount = 0
 
+	streamId := shortid.MustGenerate()
+	fileWriter.SetStreamID(streamId)
+
 	_currentBroadcast = &models.CurrentBroadcast{
+		StreamID:       streamId,
 		LatencyLevel:   data.GetStreamLatencyLevel(),
 		OutputSettings: data.GetStreamOutputVariants(),
 	}
 
 	StopOfflineCleanupTimer()
-	startOnlineCleanupTimer()
+
+	if !config.EnableReplayFeatures {
+		startOnlineCleanupTimer()
+	}
 
 	if _yp != nil {
 		go _yp.Start()
 	}
 
-	segmentPath := config.HLSStoragePath
-
 	if err := setupStorage(); err != nil {
 		log.Fatalln("failed to setup the storage", err)
 	}
 
-	go func() {
+	setupVideoComponentsForId(streamId)
-		_transcoder = transcoder.NewTranscoder()
+	setupLiveTranscoderForId(streamId, rtmpOut)
-		_transcoder.TranscoderCompleted = func(error) {
-			SetStreamAsDisconnected()
-			_transcoder = nil
-			_currentBroadcast = nil
-		}
-		_transcoder.SetStdin(rtmpOut)
-		_transcoder.Start(true)
-	}()
 
-	go webhooks.SendStreamStatusEvent(models.StreamStarted)
+	go webhooks.SendStreamStatusEvent(models.StreamStarted, streamId)
+	segmentPath := filepath.Join(config.HLSStoragePath, streamId)
 	transcoder.StartThumbnailGenerator(segmentPath, data.FindHighestVideoQualityIndex(_currentBroadcast.OutputSettings))
 
 	_ = chat.SendSystemAction("Stay tuned, the stream is **starting**!", true)
@@ -100,6 +100,7 @@ func SetStreamAsDisconnected() {
 		return
 	}
 
+	handler.StreamEnded()
 	transcoder.StopThumbnailGenerator()
 	rtmp.Disconnect()
 
@@ -118,14 +119,14 @@ func SetStreamAsDisconnected() {
 	}
 
 	for index := range _currentBroadcast.OutputSettings {
-		makeVariantIndexOffline(index, offlineFilePath, offlineFilename)
+		makeVariantIndexOffline(_currentBroadcast.StreamID, index, offlineFilePath, offlineFilename)
 	}
 
 	StartOfflineCleanupTimer()
 	stopOnlineCleanupTimer()
 	saveStats()
 
-	go webhooks.SendStreamStatusEvent(models.StreamStopped)
+	go webhooks.SendStreamStatusEvent(models.StreamStopped, _currentBroadcast.StreamID)
 }
 
 // StartOfflineCleanupTimer will fire a cleanup after n minutes being disconnected.

core/transcoder/fileWriterReceiverService.go

@@ -26,6 +26,7 @@ type FileWriterReceiverServiceCallback interface {
 // as it can send HTTP requests to this service with the results.
 type FileWriterReceiverService struct {
 	callbacks FileWriterReceiverServiceCallback
+	streamId  string
 }
 
 // SetupFileWriterReceiverService will start listening for transcoder responses.
@@ -53,6 +54,10 @@ func (s *FileWriterReceiverService) SetupFileWriterReceiverService(callbacks Fil
 	}()
 }
 
+func (s *FileWriterReceiverService) SetStreamID(streamID string) {
+	s.streamId = streamID
+}
+
 func (s *FileWriterReceiverService) uploadHandler(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "PUT" {
 		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
@@ -79,7 +84,7 @@ func (s *FileWriterReceiverService) uploadHandler(w http.ResponseWriter, r *http
 }
 
 func (s *FileWriterReceiverService) fileWritten(path string) {
-	if utils.GetRelativePathFromAbsolutePath(path) == "hls/stream.m3u8" {
+	if utils.GetRelativePathFromAbsolutePath(path) == s.streamId+"/stream.m3u8" {
 		s.callbacks.MasterPlaylistWritten(path)
 	} else if strings.HasSuffix(path, ".ts") {
 		s.callbacks.SegmentWritten(path)

core/transcoder/hlsHandler.go

@@ -1,17 +1,46 @@
 package transcoder
 
 import (
+	"github.com/owncast/owncast/config"
 	"github.com/owncast/owncast/models"
+	"github.com/owncast/owncast/replays"
+	log "github.com/sirupsen/logrus"
 )
 
 // HLSHandler gets told about available HLS playlists and segments.
 type HLSHandler struct {
-	Storage models.StorageProvider
+	Storage  models.StorageProvider
+	Recorder *replays.HLSRecorder
+}
+
+// StreamEnded is called when a stream is ended so the end time can be noted
+// in the stream's metadata.
+func (h *HLSHandler) StreamEnded() {
+	if config.EnableReplayFeatures {
+		h.Recorder.StreamEnded()
+	}
+}
+
+func (h *HLSHandler) SetStreamId(streamId string) {
+	h.Storage.SetStreamId(streamId)
+	if config.EnableReplayFeatures {
+		h.Recorder = replays.NewRecording(streamId)
+	}
 }
 
 // SegmentWritten is fired when a HLS segment is written to disk.
 func (h *HLSHandler) SegmentWritten(localFilePath string) {
-	h.Storage.SegmentWritten(localFilePath)
+	remotePath, _, err := h.Storage.SegmentWritten(localFilePath)
+	if err != nil {
+		log.Debugln(err, localFilePath)
+		return
+	}
+
+	if h.Recorder != nil {
+		h.Recorder.SegmentWritten(remotePath)
+	} else {
+		log.Debugln("No HLS recorder available to notify of segment written.")
+	}
 }
 
 // VariantPlaylistWritten is fired when a HLS variant playlist is written to disk.

core/transcoder/transcoder.go

@@ -27,12 +27,10 @@ type Transcoder struct {
 	stdin *io.PipeReader
 
 	TranscoderCompleted  func(error)
-	playlistOutputPath   string
+	StreamID             string
 	ffmpegPath           string
-	segmentIdentifier    string
 	internalListenerPort string
 	input                string
-	segmentOutputPath    string
 	variants             []HLSVariant
 
 	currentStreamOutputSettings []models.StreamOutputVariant
@@ -118,7 +116,9 @@ func (t *Transcoder) Start(shouldLog bool) {
 	if shouldLog {
 		log.Infof("Processing video using codec %s with %d output qualities configured.", t.codec.DisplayName(), len(t.variants))
 	}
-	createVariantDirectories()
+
+	// Make directory for this stream.
+	createVariantDirectories(t.StreamID)
 
 	if config.EnableDebugFeatures {
 		log.Println(command)
@@ -181,8 +181,8 @@ func (t *Transcoder) getString() string {
 		hlsOptionFlags = append(hlsOptionFlags, "append_list")
 	}
 
-	if t.segmentIdentifier == "" {
+	if t.StreamID == "" {
-		t.segmentIdentifier = shortid.MustGenerate()
+		t.StreamID = shortid.MustGenerate()
 	}
 
 	hlsEventString := ""
@@ -197,6 +197,7 @@ func (t *Transcoder) getString() string {
 	if len(hlsOptionFlags) > 0 {
 		hlsOptionsString = "-hls_flags " + strings.Join(hlsOptionFlags, "+")
 	}
+
 	ffmpegFlags := []string{
 		fmt.Sprintf(`FFREPORT=file="%s":level=32`, logging.GetTranscoderLogFilePath()),
 		t.ffmpegPath,
@@ -226,11 +227,11 @@ func (t *Transcoder) getString() string {
 		// Filenames
 		"-master_pl_name", "stream.m3u8",
 
-		"-hls_segment_filename", localListenerAddress + "/%v/stream-" + t.segmentIdentifier + "-%d.ts", // Send HLS segments back to us over HTTP
+		"-hls_segment_filename", localListenerAddress + "/" + t.StreamID + "/%v/stream-" + t.StreamID + "-%d.ts", // Send HLS segments back to us over HTTP
 		"-max_muxing_queue_size", "400", // Workaround for Too many packets error: https://trac.ffmpeg.org/ticket/6375?cversion=0
 
-		"-method PUT",                            // HLS results sent back to us will be over PUTs
+		"-method PUT", // HLS results sent back to us will be over PUTs
-		localListenerAddress + "/%v/stream.m3u8", // Send HLS playlists back to us over HTTP
+		localListenerAddress + "/" + t.StreamID + "/%v/stream.m3u8", // Send HLS playlists back to us over HTTP
 	}
 
 	return strings.Join(ffmpegFlags, " ")
@@ -272,19 +273,17 @@ func getVariantFromConfigQuality(quality models.StreamOutputVariant, index int)
 }
 
 // NewTranscoder will return a new Transcoder, populated by the config.
-func NewTranscoder() *Transcoder {
+func NewTranscoder(streamID string) *Transcoder {
 	ffmpegPath := utils.ValidatedFfmpegPath(data.GetFfMpegPath())
 
 	transcoder := new(Transcoder)
+	transcoder.StreamID = streamID
 	transcoder.ffmpegPath = ffmpegPath
 	transcoder.internalListenerPort = config.InternalHLSListenerPort
 
 	transcoder.currentStreamOutputSettings = data.GetStreamOutputVariants()
 	transcoder.currentLatencyLevel = data.GetStreamLatencyLevel()
 	transcoder.codec = getCodec(data.GetVideoCodec())
-	transcoder.segmentOutputPath = config.HLSStoragePath
-	transcoder.playlistOutputPath = config.HLSStoragePath
-
 	transcoder.input = "pipe:0" // stdin
 
 	for index, quality := range transcoder.currentStreamOutputSettings {
@@ -433,14 +432,9 @@ func (t *Transcoder) SetStdin(pipe *io.PipeReader) {
 	t.stdin = pipe
 }
 
-// SetOutputPath sets the root directory that should include playlists and video segments.
+// SetStreamID sets a unique identifier for the currently transcoding stream.
-func (t *Transcoder) SetOutputPath(output string) {
+func (t *Transcoder) SetStreamID(id string) {
-	t.segmentOutputPath = output
+	t.StreamID = id
-}
-
-// SetIdentifier enables appending a unique identifier to segment file name.
-func (t *Transcoder) SetIdentifier(output string) {
-	t.segmentIdentifier = output
 }
 
 // SetInternalHTTPPort will set the port to be used for internal communication.

core/transcoder/transcoder_nvenc_test.go

@@ -14,8 +14,7 @@ func TestFFmpegNvencCommand(t *testing.T) {
 	transcoder := new(Transcoder)
 	transcoder.ffmpegPath = filepath.Join("fake", "path", "ffmpeg")
 	transcoder.SetInput("fakecontent.flv")
-	transcoder.SetOutputPath("fakeOutput")
+	transcoder.SetStreamID("jdFsdfzGg")
-	transcoder.SetIdentifier("jdoieGg")
 	transcoder.SetInternalHTTPPort("8123")
 	transcoder.SetCodec(codec.Name())
 	transcoder.currentLatencyLevel = latencyLevel
@@ -42,7 +41,7 @@ func TestFFmpegNvencCommand(t *testing.T) {
 	cmd := transcoder.getString()
 
 	expectedLogPath := filepath.Join("data", "logs", "transcoder.log")
-	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning -hwaccel cuda -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_nvenc -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30 -tune:v:0 ll -map a:0? -c:a:0 copy -preset p3 -map v:0 -c:v:1 h264_nvenc -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24 -tune:v:1 ll -map a:0? -c:a:1 copy -preset p5 -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset p1  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1  -pix_fmt yuv420p -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/%v/stream-jdoieGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/%v/stream.m3u8`
+	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning -hwaccel cuda -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_nvenc -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30 -tune:v:0 ll -map a:0? -c:a:0 copy -preset p3 -map v:0 -c:v:1 h264_nvenc -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24 -tune:v:1 ll -map a:0? -c:a:1 copy -preset p5 -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset p1  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1  -pix_fmt yuv420p -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/jdFsdfzGg/%v/stream-jdFsdfzGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/jdFsdfzGg/%v/stream.m3u8`
 
 	if cmd != expected {
 		t.Errorf("ffmpeg command does not match expected.\nGot %s\n, want: %s", cmd, expected)

core/transcoder/transcoder_omx_test.go

@@ -14,8 +14,7 @@ func TestFFmpegOmxCommand(t *testing.T) {
 	transcoder := new(Transcoder)
 	transcoder.ffmpegPath = filepath.Join("fake", "path", "ffmpeg")
 	transcoder.SetInput("fakecontent.flv")
-	transcoder.SetOutputPath("fakeOutput")
+	transcoder.SetStreamID("jdFsdfzGg")
-	transcoder.SetIdentifier("jdFsdfzGg")
 	transcoder.SetInternalHTTPPort("8123")
 	transcoder.SetCodec(codec.Name())
 	transcoder.currentLatencyLevel = latencyLevel
@@ -42,7 +41,7 @@ func TestFFmpegOmxCommand(t *testing.T) {
 	cmd := transcoder.getString()
 
 	expectedLogPath := filepath.Join("data", "logs", "transcoder.log")
-	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning  -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_omx -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30  -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 h264_omx -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24  -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1 -tune zerolatency -pix_fmt yuv420p -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/%v/stream-jdFsdfzGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/%v/stream.m3u8`
+	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning  -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_omx -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30  -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 h264_omx -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24  -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1 -tune zerolatency -pix_fmt yuv420p -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/jdFsdfzGg/%v/stream-jdFsdfzGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/jdFsdfzGg/%v/stream.m3u8`
 
 	if cmd != expected {
 		t.Errorf("ffmpeg command does not match expected.\nGot %s\n, want: %s", cmd, expected)

core/transcoder/transcoder_vaapi_test.go

@@ -14,8 +14,7 @@ func TestFFmpegVaapiCommand(t *testing.T) {
 	transcoder := new(Transcoder)
 	transcoder.ffmpegPath = filepath.Join("fake", "path", "ffmpeg")
 	transcoder.SetInput("fakecontent.flv")
-	transcoder.SetOutputPath("fakeOutput")
+	transcoder.SetStreamID("jdFsdfzGg")
-	transcoder.SetIdentifier("jdofFGg")
 	transcoder.SetInternalHTTPPort("8123")
 	transcoder.SetCodec(codec.Name())
 	transcoder.currentLatencyLevel = latencyLevel
@@ -42,7 +41,7 @@ func TestFFmpegVaapiCommand(t *testing.T) {
 	cmd := transcoder.getString()
 
 	expectedLogPath := filepath.Join("data", "logs", "transcoder.log")
-	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning -hwaccel vaapi -hwaccel_output_format vaapi -vaapi_device /dev/dri/renderD128 -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_vaapi -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30  -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 h264_vaapi -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24  -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1  -pix_fmt vaapi_vld -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/%v/stream-jdofFGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/%v/stream.m3u8`
+	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning -hwaccel vaapi -hwaccel_output_format vaapi -vaapi_device /dev/dri/renderD128 -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_vaapi -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30  -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 h264_vaapi -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24  -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1  -pix_fmt vaapi_vld -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/jdFsdfzGg/%v/stream-jdFsdfzGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/jdFsdfzGg/%v/stream.m3u8`
 
 	if cmd != expected {
 		t.Errorf("ffmpeg command does not match expected.\nGot %s\n, want: %s", cmd, expected)

core/transcoder/transcoder_videotoolbox_test.go

@@ -14,8 +14,7 @@ func TestFFmpegVideoToolboxCommand(t *testing.T) {
 	transcoder := new(Transcoder)
 	transcoder.ffmpegPath = filepath.Join("fake", "path", "ffmpeg")
 	transcoder.SetInput("fakecontent.flv")
-	transcoder.SetOutputPath("fakeOutput")
+	transcoder.SetStreamID("jdFsdfzGg")
-	transcoder.SetIdentifier("jdFsdfzGg")
 	transcoder.SetInternalHTTPPort("8123")
 	transcoder.SetCodec(codec.Name())
 	transcoder.currentLatencyLevel = latencyLevel
@@ -42,7 +41,7 @@ func TestFFmpegVideoToolboxCommand(t *testing.T) {
 	cmd := transcoder.getString()
 
 	expectedLogPath := filepath.Join("data", "logs", "transcoder.log")
-	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning  -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_videotoolbox -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30 -realtime true -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 h264_videotoolbox -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24  -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1  -pix_fmt nv12 -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/%v/stream-jdFsdfzGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/%v/stream.m3u8`
+	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning  -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 h264_videotoolbox -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30 -realtime true -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 h264_videotoolbox -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24  -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1  -pix_fmt nv12 -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/jdFsdfzGg/%v/stream-jdFsdfzGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/jdFsdfzGg/%v/stream.m3u8`
 
 	if cmd != expected {
 		t.Errorf("ffmpeg command does not match expected.\nGot %s\n, want: %s", cmd, expected)

core/transcoder/transcoder_x264_test.go

@@ -14,8 +14,7 @@ func TestFFmpegx264Command(t *testing.T) {
 	transcoder := new(Transcoder)
 	transcoder.ffmpegPath = filepath.Join("fake", "path", "ffmpeg")
 	transcoder.SetInput("fakecontent.flv")
-	transcoder.SetOutputPath("fakeOutput")
+	transcoder.SetStreamID("jdFsdfzGg")
-	transcoder.SetIdentifier("jdofFGg")
 	transcoder.SetInternalHTTPPort("8123")
 	transcoder.SetCodec(codec.Name())
 	transcoder.currentLatencyLevel = latencyLevel
@@ -42,7 +41,7 @@ func TestFFmpegx264Command(t *testing.T) {
 	cmd := transcoder.getString()
 
 	expectedLogPath := filepath.Join("data", "logs", "transcoder.log")
-	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning  -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 libx264 -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30 -x264-params:v:0 "scenecut=0:open_gop=0" -bufsize:v:0 1088k -profile:v:0 high -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 libx264 -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24 -x264-params:v:1 "scenecut=0:open_gop=0" -bufsize:v:1 3572k -profile:v:1 high -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1 -tune zerolatency -pix_fmt yuv420p -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/%v/stream-jdofFGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/%v/stream.m3u8`
+	expected := `FFREPORT=file="` + expectedLogPath + `":level=32 ` + transcoder.ffmpegPath + ` -hide_banner -loglevel warning  -fflags +genpts -flags +cgop -i  fakecontent.flv  -map v:0 -c:v:0 libx264 -b:v:0 1008k -maxrate:v:0 1088k -g:v:0 90 -keyint_min:v:0 90 -r:v:0 30 -x264-params:v:0 "scenecut=0:open_gop=0" -bufsize:v:0 1088k -profile:v:0 high -map a:0? -c:a:0 copy -preset veryfast -map v:0 -c:v:1 libx264 -b:v:1 3308k -maxrate:v:1 3572k -g:v:1 72 -keyint_min:v:1 72 -r:v:1 24 -x264-params:v:1 "scenecut=0:open_gop=0" -bufsize:v:1 3572k -profile:v:1 high -map a:0? -c:a:1 copy -preset fast -map v:0 -c:v:2 copy -map a:0? -c:a:2 copy -preset ultrafast  -var_stream_map "v:0,a:0 v:1,a:1 v:2,a:2 " -f hls -hls_time 3 -hls_list_size 10 -hls_flags program_date_time+independent_segments+omit_endlist  -segment_format_options mpegts_flags=mpegts_copyts=1 -tune zerolatency -pix_fmt yuv420p -sc_threshold 0 -master_pl_name stream.m3u8 -hls_segment_filename http://127.0.0.1:8123/jdFsdfzGg/%v/stream-jdFsdfzGg-%d.ts -max_muxing_queue_size 400 -method PUT http://127.0.0.1:8123/jdFsdfzGg/%v/stream.m3u8`
 
 	if cmd != expected {
 		t.Errorf("ffmpeg command does not match expected.\nGot %s\n, want: %s", cmd, expected)

core/transcoder/utils.go

@@ -96,13 +96,13 @@ func handleTranscoderMessage(message string) {
 	_lastTranscoderLogMessage = message
 }
 
-func createVariantDirectories() {
+func createVariantDirectories(streamID string) {
 	// Create private hls data dirs
-	utils.CleanupDirectory(config.HLSStoragePath)
+	utils.CleanupDirectory(config.HLSStoragePath, config.EnableReplayFeatures)
 
 	if len(data.GetStreamOutputVariants()) != 0 {
 		for index := range data.GetStreamOutputVariants() {
-			if err := os.MkdirAll(path.Join(config.HLSStoragePath, strconv.Itoa(index)), 0o750); err != nil {
+			if err := os.MkdirAll(path.Join(config.HLSStoragePath, streamID, strconv.Itoa(index)), 0o750); err != nil {
 				log.Fatalln(err)
 			}
 		}

core/video.go

@@ -0,0 +1,26 @@
+package core
+
+import (
+	"io"
+
+	"github.com/owncast/owncast/core/transcoder"
+)
+
+func setupVideoComponentsForId(streamId string) {
+}
+
+func setupLiveTranscoderForId(streamId string, rtmpOut *io.PipeReader) {
+	_storage.SetStreamId(streamId)
+	handler.SetStreamId(streamId)
+
+	go func() {
+		_transcoder = transcoder.NewTranscoder(streamId)
+		_transcoder.TranscoderCompleted = func(error) {
+			SetStreamAsDisconnected()
+			_transcoder = nil
+			_currentBroadcast = nil
+		}
+		_transcoder.SetStdin(rtmpOut)
+		_transcoder.Start(true)
+	}()
+}

core/webhooks/stream.go

@@ -9,11 +9,11 @@ import (
 )
 
 // SendStreamStatusEvent will send all webhook destinations the current stream status.
-func SendStreamStatusEvent(eventType models.EventType) {
+func SendStreamStatusEvent(eventType models.EventType, streamID string) {
-	sendStreamStatusEvent(eventType, shortid.MustGenerate(), time.Now())
+	sendStreamStatusEvent(eventType, shortid.MustGenerate(), streamID, time.Now())
 }
 
-func sendStreamStatusEvent(eventType models.EventType, id string, timestamp time.Time) {
+func sendStreamStatusEvent(eventType models.EventType, id, streamID string, timestamp time.Time) {
 	SendEventToWebhooks(WebhookEvent{
 		Type: eventType,
 		EventData: map[string]interface{}{
@@ -23,6 +23,7 @@ func sendStreamStatusEvent(eventType models.EventType, id string, timestamp time
 			"streamTitle": data.GetStreamTitle(),
 			"status":      getStatus(),
 			"timestamp":   timestamp,
+			"streamID":    streamID,
 		},
 	})
 }

core/webhooks/stream_test.go

@@ -14,14 +14,17 @@ func TestSendStreamStatusEvent(t *testing.T) {
 	data.SetServerSummary("my server where I stream")
 	data.SetStreamTitle("my stream")
 
+	streamID := "test-stream-id"
+
 	checkPayload(t, models.StreamStarted, func() {
-		sendStreamStatusEvent(events.StreamStarted, "id", time.Unix(72, 6).UTC())
+		sendStreamStatusEvent(events.StreamStarted, "id", streamID, time.Unix(72, 6).UTC())
 	}, `{
 		"id": "id",
 		"name": "my server",
 		"streamTitle": "my stream",
 		"summary": "my server where I stream",
 		"timestamp": "1970-01-01T00:01:12.000000006Z",
+		"streamID": "test-stream-id",
 		"status": {
 			"lastConnectTime": null,
 			"lastDisconnectTime": null,

db/db.go

@@ -1,6 +1,6 @@
 // Code generated by sqlc. DO NOT EDIT.
 // versions:
-//   sqlc v1.15.0
+//   sqlc v1.19.1
 
 package db
 

db/models.go

@@ -1,6 +1,6 @@
 // Code generated by sqlc. DO NOT EDIT.
 // versions:
-//   sqlc v1.15.0
+//   sqlc v1.19.1
 
 package db
 
@@ -72,6 +72,23 @@ type Notification struct {
 	CreatedAt   sql.NullTime
 }
 
+type ReplayClip struct {
+	ID                string
+	StreamID          string
+	ClippedBy         sql.NullString
+	ClipTitle         sql.NullString
+	RelativeStartTime sql.NullFloat64
+	RelativeEndTime   sql.NullFloat64
+	Timestamp         sql.NullTime
+}
+
+type Stream struct {
+	ID          string
+	StreamTitle sql.NullString
+	StartTime   sql.NullTime
+	EndTime     sql.NullTime
+}
+
 type User struct {
 	ID              string
 	DisplayName     string
@@ -91,3 +108,25 @@ type UserAccessToken struct {
 	UserID    string
 	Timestamp time.Time
 }
+
+type VideoSegment struct {
+	ID                    string
+	StreamID              string
+	OutputConfigurationID string
+	Path                  string
+	RelativeTimestamp     float32
+	Timestamp             sql.NullTime
+}
+
+type VideoSegmentOutputConfiguration struct {
+	ID               string
+	VariantID        string
+	Name             string
+	StreamID         string
+	SegmentDuration  int32
+	Bitrate          int32
+	Framerate        int32
+	ResolutionWidth  sql.NullInt32
+	ResolutionHeight sql.NullInt32
+	Timestamp        sql.NullTime
+}

db/query.sql

@@ -108,3 +108,61 @@ UPDATE users SET display_name = $1, previous_names = previous_names || $2, namec
 
 -- name: ChangeDisplayColor :exec
 UPDATE users SET display_color = $1 WHERE id = $2;
+
+-- Recording and clip related queries.
+
+-- name: GetStreams :many
+SELECT id, stream_title, start_time, end_time FROM streams ORDER BY start_time DESC;
+
+-- name: GetStreamById :one
+SELECT id, stream_title, start_time, end_time FROM streams WHERE id = $1 LIMIT 1;
+
+-- name: GetOutputConfigurationsForStreamId :many
+SELECT id, stream_id, variant_id, name, segment_duration, bitrate, framerate, resolution_width, resolution_height FROM video_segment_output_configuration WHERE stream_id = $1;
+
+-- name: GetOutputConfigurationForId :one
+SELECT id, stream_id, variant_id, name, segment_duration, bitrate, framerate, resolution_width, resolution_height FROM video_segment_output_configuration WHERE id = $1;
+
+-- name: GetSegmentsForOutputId :many
+SELECT id, stream_id, output_configuration_id, path, timestamp FROM video_segments WHERE output_configuration_id = $1 ORDER BY timestamp ASC;
+
+-- name: GetSegmentsForOutputIdAndWindow :many
+SELECT id, stream_id, output_configuration_id, path, relative_timestamp, timestamp FROM video_segments WHERE output_configuration_id = $1 AND (cast ( relative_timestamp as int ) - ( relative_timestamp < cast ( relative_timestamp as int ))) >= @start_seconds::REAL AND (cast ( relative_timestamp as int ) + ( relative_timestamp > cast ( relative_timestamp as int ))) <= @end_seconds::REAL ORDER BY relative_timestamp ASC;
+
+-- name: InsertStream :exec
+INSERT INTO streams (id, stream_title, start_time, end_time) VALUES($1, $2, $3, $4);
+
+-- name: InsertOutputConfiguration :exec
+INSERT INTO video_segment_output_configuration (id, variant_id, stream_id, name, segment_duration, bitrate, framerate, resolution_width, resolution_height, timestamp) VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10);
+
+-- name: InsertSegment :exec
+INSERT INTO video_segments (id, stream_id, output_configuration_id, path, relative_timestamp, timestamp) VALUES($1, $2, $3, $4, $5, $6);
+
+-- name: SetStreamEnded :exec
+UPDATE streams SET end_time = $1 WHERE id = $2;
+
+-- name: InsertClip :exec
+INSERT INTO replay_clips (id, stream_id, clip_title, relative_start_time, relative_end_time, timestamp) VALUES($1, $2, $3, $4, $5, $6);
+
+-- name: GetAllClips :many
+SELECT rc.id AS id, rc.clip_title, rc.stream_id, rc.relative_start_time, rc.relative_end_time, (rc.relative_end_time - rc.relative_start_time) AS duration_seconds, rc.timestamp, s.stream_title AS stream_title
+	FROM replay_clips rc
+	JOIN streams s ON rc.stream_id = s.id
+	ORDER BY timestamp DESC;
+
+-- name: GetAllClipsForStream :many
+SELECT rc.id AS clip_id, rc.stream_id, rc.clipped_by, rc.clip_title, rc.relative_start_time, rc.relative_end_time, rc.timestamp,
+	s.id AS stream_id, s.stream_title AS stream_title
+	FROM replay_clips rc
+	JOIN streams s ON rc.stream_id = s.id
+	WHERE rc.stream_id = $1
+	ORDER BY timestamp DESC;
+
+-- name: GetClip :one
+SELECT id AS clip_id, stream_id, clipped_by, clip_title, timestamp AS clip_timestamp, relative_start_time, relative_end_time FROM replay_clips WHERE id = $1;
+
+-- name: GetFinalSegmentForStream :one
+SELECT id, stream_id, output_configuration_id, path, relative_timestamp, timestamp FROM video_segments WHERE stream_id = $1 ORDER BY relative_timestamp DESC LIMIT 1;
+
+-- name: FixUnfinishedStreams :exec
+UPDATE streams SET end_time = (SELECT timestamp FROM video_segments WHERE stream_id = streams.id) WHERE end_time IS NULL;

db/query.sql.go

@@ -205,6 +205,148 @@ func (q *Queries) DoesInboundActivityExist(ctx context.Context, arg DoesInboundA
 	return count, err
 }
 
+const fixUnfinishedStreams = `-- name: FixUnfinishedStreams :exec
+UPDATE streams SET end_time = (SELECT timestamp FROM video_segments WHERE stream_id = streams.id) WHERE end_time IS NULL
+`
+
+func (q *Queries) FixUnfinishedStreams(ctx context.Context) error {
+	_, err := q.db.ExecContext(ctx, fixUnfinishedStreams)
+	return err
+}
+
+const getAllClips = `-- name: GetAllClips :many
+SELECT rc.id AS id, rc.clip_title, rc.stream_id, rc.relative_start_time, rc.relative_end_time, (rc.relative_end_time - rc.relative_start_time) AS duration_seconds, rc.timestamp, s.stream_title AS stream_title
+	FROM replay_clips rc
+	JOIN streams s ON rc.stream_id = s.id
+	ORDER BY timestamp DESC
+`
+
+type GetAllClipsRow struct {
+	ID                string
+	ClipTitle         sql.NullString
+	StreamID          string
+	RelativeStartTime sql.NullFloat64
+	RelativeEndTime   sql.NullFloat64
+	DurationSeconds   int32
+	Timestamp         sql.NullTime
+	StreamTitle       sql.NullString
+}
+
+func (q *Queries) GetAllClips(ctx context.Context) ([]GetAllClipsRow, error) {
+	rows, err := q.db.QueryContext(ctx, getAllClips)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetAllClipsRow
+	for rows.Next() {
+		var i GetAllClipsRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.ClipTitle,
+			&i.StreamID,
+			&i.RelativeStartTime,
+			&i.RelativeEndTime,
+			&i.DurationSeconds,
+			&i.Timestamp,
+			&i.StreamTitle,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const getAllClipsForStream = `-- name: GetAllClipsForStream :many
+SELECT rc.id AS clip_id, rc.stream_id, rc.clipped_by, rc.clip_title, rc.relative_start_time, rc.relative_end_time, rc.timestamp,
+	s.id AS stream_id, s.stream_title AS stream_title
+	FROM replay_clips rc
+	JOIN streams s ON rc.stream_id = s.id
+	WHERE rc.stream_id = $1
+	ORDER BY timestamp DESC
+`
+
+type GetAllClipsForStreamRow struct {
+	ClipID            string
+	StreamID          string
+	ClippedBy         sql.NullString
+	ClipTitle         sql.NullString
+	RelativeStartTime sql.NullFloat64
+	RelativeEndTime   sql.NullFloat64
+	Timestamp         sql.NullTime
+	StreamID_2        string
+	StreamTitle       sql.NullString
+}
+
+func (q *Queries) GetAllClipsForStream(ctx context.Context, streamID string) ([]GetAllClipsForStreamRow, error) {
+	rows, err := q.db.QueryContext(ctx, getAllClipsForStream, streamID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetAllClipsForStreamRow
+	for rows.Next() {
+		var i GetAllClipsForStreamRow
+		if err := rows.Scan(
+			&i.ClipID,
+			&i.StreamID,
+			&i.ClippedBy,
+			&i.ClipTitle,
+			&i.RelativeStartTime,
+			&i.RelativeEndTime,
+			&i.Timestamp,
+			&i.StreamID_2,
+			&i.StreamTitle,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const getClip = `-- name: GetClip :one
+SELECT id AS clip_id, stream_id, clipped_by, clip_title, timestamp AS clip_timestamp, relative_start_time, relative_end_time FROM replay_clips WHERE id = $1
+`
+
+type GetClipRow struct {
+	ClipID            string
+	StreamID          string
+	ClippedBy         sql.NullString
+	ClipTitle         sql.NullString
+	ClipTimestamp     sql.NullTime
+	RelativeStartTime sql.NullFloat64
+	RelativeEndTime   sql.NullFloat64
+}
+
+func (q *Queries) GetClip(ctx context.Context, id string) (GetClipRow, error) {
+	row := q.db.QueryRowContext(ctx, getClip, id)
+	var i GetClipRow
+	err := row.Scan(
+		&i.ClipID,
+		&i.StreamID,
+		&i.ClippedBy,
+		&i.ClipTitle,
+		&i.ClipTimestamp,
+		&i.RelativeStartTime,
+		&i.RelativeEndTime,
+	)
+	return i, err
+}
+
 const getFederationFollowerApprovalRequests = `-- name: GetFederationFollowerApprovalRequests :many
 SELECT iri, inbox, name, username, image, created_at FROM ap_followers WHERE approved_at IS null AND disabled_at is null
 `
@@ -296,6 +438,24 @@ func (q *Queries) GetFederationFollowersWithOffset(ctx context.Context, arg GetF
 	return items, nil
 }
 
+const getFinalSegmentForStream = `-- name: GetFinalSegmentForStream :one
+SELECT id, stream_id, output_configuration_id, path, relative_timestamp, timestamp FROM video_segments WHERE stream_id = $1 ORDER BY relative_timestamp DESC LIMIT 1
+`
+
+func (q *Queries) GetFinalSegmentForStream(ctx context.Context, streamID string) (VideoSegment, error) {
+	row := q.db.QueryRowContext(ctx, getFinalSegmentForStream, streamID)
+	var i VideoSegment
+	err := row.Scan(
+		&i.ID,
+		&i.StreamID,
+		&i.OutputConfigurationID,
+		&i.Path,
+		&i.RelativeTimestamp,
+		&i.Timestamp,
+	)
+	return i, err
+}
+
 const getFollowerByIRI = `-- name: GetFollowerByIRI :one
 SELECT iri, inbox, name, username, image, request, request_object, created_at, approved_at, disabled_at FROM ap_followers WHERE iri = $1
 `
@@ -541,6 +701,88 @@ func (q *Queries) GetOutboxWithOffset(ctx context.Context, arg GetOutboxWithOffs
 	return items, nil
 }
 
+const getOutputConfigurationForId = `-- name: GetOutputConfigurationForId :one
+SELECT id, stream_id, variant_id, name, segment_duration, bitrate, framerate, resolution_width, resolution_height FROM video_segment_output_configuration WHERE id = $1
+`
+
+type GetOutputConfigurationForIdRow struct {
+	ID               string
+	StreamID         string
+	VariantID        string
+	Name             string
+	SegmentDuration  int32
+	Bitrate          int32
+	Framerate        int32
+	ResolutionWidth  sql.NullInt32
+	ResolutionHeight sql.NullInt32
+}
+
+func (q *Queries) GetOutputConfigurationForId(ctx context.Context, id string) (GetOutputConfigurationForIdRow, error) {
+	row := q.db.QueryRowContext(ctx, getOutputConfigurationForId, id)
+	var i GetOutputConfigurationForIdRow
+	err := row.Scan(
+		&i.ID,
+		&i.StreamID,
+		&i.VariantID,
+		&i.Name,
+		&i.SegmentDuration,
+		&i.Bitrate,
+		&i.Framerate,
+		&i.ResolutionWidth,
+		&i.ResolutionHeight,
+	)
+	return i, err
+}
+
+const getOutputConfigurationsForStreamId = `-- name: GetOutputConfigurationsForStreamId :many
+SELECT id, stream_id, variant_id, name, segment_duration, bitrate, framerate, resolution_width, resolution_height FROM video_segment_output_configuration WHERE stream_id = $1
+`
+
+type GetOutputConfigurationsForStreamIdRow struct {
+	ID               string
+	StreamID         string
+	VariantID        string
+	Name             string
+	SegmentDuration  int32
+	Bitrate          int32
+	Framerate        int32
+	ResolutionWidth  sql.NullInt32
+	ResolutionHeight sql.NullInt32
+}
+
+func (q *Queries) GetOutputConfigurationsForStreamId(ctx context.Context, streamID string) ([]GetOutputConfigurationsForStreamIdRow, error) {
+	rows, err := q.db.QueryContext(ctx, getOutputConfigurationsForStreamId, streamID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetOutputConfigurationsForStreamIdRow
+	for rows.Next() {
+		var i GetOutputConfigurationsForStreamIdRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.StreamID,
+			&i.VariantID,
+			&i.Name,
+			&i.SegmentDuration,
+			&i.Bitrate,
+			&i.Framerate,
+			&i.ResolutionWidth,
+			&i.ResolutionHeight,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getRejectedAndBlockedFollowers = `-- name: GetRejectedAndBlockedFollowers :many
 SELECT iri, name, username, image, created_at, disabled_at FROM ap_followers WHERE disabled_at is not null
 `
@@ -584,6 +826,137 @@ func (q *Queries) GetRejectedAndBlockedFollowers(ctx context.Context) ([]GetReje
 	return items, nil
 }
 
+const getSegmentsForOutputId = `-- name: GetSegmentsForOutputId :many
+SELECT id, stream_id, output_configuration_id, path, timestamp FROM video_segments WHERE output_configuration_id = $1 ORDER BY timestamp ASC
+`
+
+type GetSegmentsForOutputIdRow struct {
+	ID                    string
+	StreamID              string
+	OutputConfigurationID string
+	Path                  string
+	Timestamp             sql.NullTime
+}
+
+func (q *Queries) GetSegmentsForOutputId(ctx context.Context, outputConfigurationID string) ([]GetSegmentsForOutputIdRow, error) {
+	rows, err := q.db.QueryContext(ctx, getSegmentsForOutputId, outputConfigurationID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetSegmentsForOutputIdRow
+	for rows.Next() {
+		var i GetSegmentsForOutputIdRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.StreamID,
+			&i.OutputConfigurationID,
+			&i.Path,
+			&i.Timestamp,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const getSegmentsForOutputIdAndWindow = `-- name: GetSegmentsForOutputIdAndWindow :many
+SELECT id, stream_id, output_configuration_id, path, relative_timestamp, timestamp FROM video_segments WHERE output_configuration_id = $1 AND (cast ( relative_timestamp as int ) - ( relative_timestamp < cast ( relative_timestamp as int ))) >= $2::REAL AND (cast ( relative_timestamp as int ) + ( relative_timestamp > cast ( relative_timestamp as int ))) <= $3::REAL ORDER BY relative_timestamp ASC
+`
+
+type GetSegmentsForOutputIdAndWindowParams struct {
+	OutputConfigurationID string
+	StartSeconds          float32
+	EndSeconds            float32
+}
+
+func (q *Queries) GetSegmentsForOutputIdAndWindow(ctx context.Context, arg GetSegmentsForOutputIdAndWindowParams) ([]VideoSegment, error) {
+	rows, err := q.db.QueryContext(ctx, getSegmentsForOutputIdAndWindow, arg.OutputConfigurationID, arg.StartSeconds, arg.EndSeconds)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []VideoSegment
+	for rows.Next() {
+		var i VideoSegment
+		if err := rows.Scan(
+			&i.ID,
+			&i.StreamID,
+			&i.OutputConfigurationID,
+			&i.Path,
+			&i.RelativeTimestamp,
+			&i.Timestamp,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
+const getStreamById = `-- name: GetStreamById :one
+SELECT id, stream_title, start_time, end_time FROM streams WHERE id = $1 LIMIT 1
+`
+
+func (q *Queries) GetStreamById(ctx context.Context, id string) (Stream, error) {
+	row := q.db.QueryRowContext(ctx, getStreamById, id)
+	var i Stream
+	err := row.Scan(
+		&i.ID,
+		&i.StreamTitle,
+		&i.StartTime,
+		&i.EndTime,
+	)
+	return i, err
+}
+
+const getStreams = `-- name: GetStreams :many
+
+SELECT id, stream_title, start_time, end_time FROM streams ORDER BY start_time DESC
+`
+
+// Recording and clip related queries.
+func (q *Queries) GetStreams(ctx context.Context) ([]Stream, error) {
+	rows, err := q.db.QueryContext(ctx, getStreams)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []Stream
+	for rows.Next() {
+		var i Stream
+		if err := rows.Scan(
+			&i.ID,
+			&i.StreamTitle,
+			&i.StartTime,
+			&i.EndTime,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getUserByAccessToken = `-- name: GetUserByAccessToken :one
 SELECT users.id, display_name, display_color, users.created_at, disabled_at, previous_names, namechanged_at, authenticated_at, scopes FROM users, user_access_tokens WHERE token = $1 AND users.id = user_id
 `
@@ -666,6 +1039,110 @@ func (q *Queries) GetUserDisplayNameByToken(ctx context.Context, token string) (
 	return display_name, err
 }
 
+const insertClip = `-- name: InsertClip :exec
+INSERT INTO replay_clips (id, stream_id, clip_title, relative_start_time, relative_end_time, timestamp) VALUES($1, $2, $3, $4, $5, $6)
+`
+
+type InsertClipParams struct {
+	ID                string
+	StreamID          string
+	ClipTitle         sql.NullString
+	RelativeStartTime sql.NullFloat64
+	RelativeEndTime   sql.NullFloat64
+	Timestamp         sql.NullTime
+}
+
+func (q *Queries) InsertClip(ctx context.Context, arg InsertClipParams) error {
+	_, err := q.db.ExecContext(ctx, insertClip,
+		arg.ID,
+		arg.StreamID,
+		arg.ClipTitle,
+		arg.RelativeStartTime,
+		arg.RelativeEndTime,
+		arg.Timestamp,
+	)
+	return err
+}
+
+const insertOutputConfiguration = `-- name: InsertOutputConfiguration :exec
+INSERT INTO video_segment_output_configuration (id, variant_id, stream_id, name, segment_duration, bitrate, framerate, resolution_width, resolution_height, timestamp) VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+`
+
+type InsertOutputConfigurationParams struct {
+	ID               string
+	VariantID        string
+	StreamID         string
+	Name             string
+	SegmentDuration  int32
+	Bitrate          int32
+	Framerate        int32
+	ResolutionWidth  sql.NullInt32
+	ResolutionHeight sql.NullInt32
+	Timestamp        sql.NullTime
+}
+
+func (q *Queries) InsertOutputConfiguration(ctx context.Context, arg InsertOutputConfigurationParams) error {
+	_, err := q.db.ExecContext(ctx, insertOutputConfiguration,
+		arg.ID,
+		arg.VariantID,
+		arg.StreamID,
+		arg.Name,
+		arg.SegmentDuration,
+		arg.Bitrate,
+		arg.Framerate,
+		arg.ResolutionWidth,
+		arg.ResolutionHeight,
+		arg.Timestamp,
+	)
+	return err
+}
+
+const insertSegment = `-- name: InsertSegment :exec
+INSERT INTO video_segments (id, stream_id, output_configuration_id, path, relative_timestamp, timestamp) VALUES($1, $2, $3, $4, $5, $6)
+`
+
+type InsertSegmentParams struct {
+	ID                    string
+	StreamID              string
+	OutputConfigurationID string
+	Path                  string
+	RelativeTimestamp     float32
+	Timestamp             sql.NullTime
+}
+
+func (q *Queries) InsertSegment(ctx context.Context, arg InsertSegmentParams) error {
+	_, err := q.db.ExecContext(ctx, insertSegment,
+		arg.ID,
+		arg.StreamID,
+		arg.OutputConfigurationID,
+		arg.Path,
+		arg.RelativeTimestamp,
+		arg.Timestamp,
+	)
+	return err
+}
+
+const insertStream = `-- name: InsertStream :exec
+INSERT INTO streams (id, stream_title, start_time, end_time) VALUES($1, $2, $3, $4)
+`
+
+type InsertStreamParams struct {
+	ID          string
+	StreamTitle sql.NullString
+	StartTime   sql.NullTime
+	EndTime     sql.NullTime
+}
+
+func (q *Queries) InsertStream(ctx context.Context, arg InsertStreamParams) error {
+	_, err := q.db.ExecContext(ctx, insertStream,
+		arg.ID,
+		arg.StreamTitle,
+		arg.StartTime,
+		arg.EndTime,
+	)
+	return err
+}
+
 const isDisplayNameAvailable = `-- name: IsDisplayNameAvailable :one
 SELECT count(*) FROM users WHERE display_name = $1 AND ( type='API' OR authenticated_at IS NOT NULL ) AND disabled_at IS NULL
 `
@@ -748,6 +1225,20 @@ func (q *Queries) SetAccessTokenToOwner(ctx context.Context, arg SetAccessTokenT
 	return err
 }
 
+const setStreamEnded = `-- name: SetStreamEnded :exec
+UPDATE streams SET end_time = $1 WHERE id = $2
+`
+
+type SetStreamEndedParams struct {
+	EndTime sql.NullTime
+	ID      string
+}
+
+func (q *Queries) SetStreamEnded(ctx context.Context, arg SetStreamEndedParams) error {
+	_, err := q.db.ExecContext(ctx, setStreamEnded, arg.EndTime, arg.ID)
+	return err
+}
+
 const setUserAsAuthenticated = `-- name: SetUserAsAuthenticated :exec
 UPDATE users SET authenticated_at = CURRENT_TIMESTAMP WHERE id = $1
 `

db/schema.sql

@@ -97,3 +97,61 @@ CREATE TABLE IF NOT EXISTS messages (
 	CREATE INDEX user_id ON messages (user_id);
 	CREATE INDEX hidden_at ON messages (hidden_at);
 	CREATE INDEX timestamp ON messages (timestamp);
+
+-- Record the high level details of each stream.
+CREATE TABLE IF NOT EXISTS streams (
+	"id" string NOT NULL PRIMARY KEY,
+	"stream_title" TEXT,
+	"start_time" DATE,
+	"end_time" DATE,
+	PRIMARY KEY (id)
+);
+CREATE INDEX streams_id ON streams (id);
+CREATE INDEX streams_start_time ON streams (start_time);
+CREATE INDEX streams_start_end_time ON streams (start_time,end_time);
+
+-- Record the output configuration of a stream.
+CREATE TABLE IF NOT EXISTS video_segment_output_configuration (
+	"id" string NOT NULL PRIMARY KEY,
+	"variant_id" string NOT NULL,
+	"name" string NOT NULL,
+	"stream_id" string NOT NULL,
+	"segment_duration" INTEGER NOT NULL,
+	"bitrate" INTEGER NOT NULL,
+	"framerate" INTEGER NOT NULL,
+	"resolution_width" INTEGER,
+	"resolution_height" INTEGER,
+	"timestamp" DATE,
+	PRIMARY KEY (id)
+);
+	CREATE INDEX video_segment_output_configuration_stream_id ON video_segment_output_configuration (stream_id);
+
+-- Support querying all segments for a single stream as well
+-- as segments for a time window.
+CREATE TABLE IF NOT EXISTS video_segments (
+	"id" string NOT NULL PRIMARY KEY,
+	"stream_id" string NOT NULL,
+	"output_configuration_id" string NOT NULL,
+	"path" TEXT NOT NULL,
+	"relative_timestamp" REAL NOT NULL,
+	"timestamp" DATE,
+	PRIMARY KEY (id)
+);
+	CREATE INDEX video_segments_stream_id ON video_segments (stream_id);
+	CREATE INDEX video_segments_stream_id_timestamp ON video_segments (stream_id,timestamp);
+
+-- Record the details of a replayable clip.
+CREATE TABLE IF NOT EXISTS replay_clips (
+	"id" string NOT NULL,
+	"stream_id" string NOT NULL,
+	"clipped_by" string,
+	"clip_title" TEXT,
+	"relative_start_time" REAL,
+	"relative_end_time" REAL,
+	"timestamp" DATE,
+	PRIMARY KEY (id),
+  FOREIGN KEY(stream_id) REFERENCES streams(id)
+);
+CREATE INDEX clip_id ON replay_clips (id);
+CREATE INDEX clip_stream_id ON replay_clips (stream_id);
+CREATE INDEX clip_start_end_time ON replay_clips (start_time,end_time);

main.go

@@ -28,6 +28,7 @@ var (
 	webServerPortOverride = flag.String("webserverport", "", "Force the web server to listen on a specific port")
 	webServerIPOverride   = flag.String("webserverip", "", "Force web server to listen on this IP address")
 	rtmpPortOverride      = flag.Int("rtmpport", 0, "Set listen port for the RTMP server")
+	enableReplayFeatures  = flag.Bool("enableReplayFeatures", false, "Enable experimental replay features")
 )
 
 // nolint:cyclop
@@ -42,6 +43,8 @@ func main() {
 		config.BackupDirectory = *backupDirectory
 	}
 
+	config.EnableReplayFeatures = *enableReplayFeatures
+
 	// Create the data directory if needed
 	if !utils.DoesFileExists("data") {
 		if err := os.Mkdir("./data", 0o700); err != nil {

models/currentBroadcast.go

@@ -2,6 +2,7 @@ package models
 
 // CurrentBroadcast represents the configuration associated with the currently active stream.
 type CurrentBroadcast struct {
+	StreamID       string                `json:"streamId"`
 	OutputSettings []StreamOutputVariant `json:"outputSettings"`
 	LatencyLevel   LatencyLevel          `json:"latencyLevel"`
 }

models/flexibledate.go

@@ -0,0 +1,63 @@
+package models
+
+import (
+	"database/sql"
+	"errors"
+	"time"
+)
+
+type FlexibleDate struct {
+	time.Time
+}
+
+func (self *FlexibleDate) UnmarshalJSON(b []byte) (err error) {
+	s := string(b)
+
+	// Get rid of the quotes "" around the value.
+	s = s[1 : len(s)-1]
+
+	result, err := FlexibleDateParse(s)
+	if err != nil {
+		return err
+	}
+
+	self.Time = result
+
+	return
+}
+
+// FlexibleDateParse is a convinience function to parse a date that could be
+// a string, a time.Time, or a sql.NullTime.
+func FlexibleDateParse(date interface{}) (time.Time, error) {
+	// If it's within a sql.NullTime wrapper, return the time from that.
+	nulltime, ok := date.(sql.NullTime)
+	if ok {
+		return nulltime.Time, nil
+	}
+
+	// Parse as string
+	datestring, ok := date.(string)
+	if ok {
+		t, err := time.Parse(time.RFC3339Nano, datestring)
+		if err == nil {
+			return t, nil
+		}
+
+		t, err = time.Parse("2006-01-02T15:04:05.999999999Z0700", datestring)
+		if err == nil {
+			return t, nil
+		}
+
+		t, err = time.Parse("2006-01-02 15:04:05.999999999-07:00", datestring)
+		if err == nil {
+			return t, nil
+		}
+	}
+
+	dateobject, ok := date.(time.Time)
+	if ok {
+		return dateobject, nil
+	}
+
+	return time.Time{}, errors.New("unable to parse date")
+}

models/flexibledate_test.go

@@ -0,0 +1,42 @@
+package models
+
+import (
+	"database/sql"
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+func TestFlexibleDateParsing(t *testing.T) {
+	type testJson struct {
+		Testdate FlexibleDate `json:"testdate"`
+	}
+
+	nullTime := sql.NullTime{Time: time.Unix(1591614434, 0), Valid: true}
+	testNullTime, err := FlexibleDateParse(nullTime)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if testNullTime.Unix() != nullTime.Time.Unix() {
+		t.Errorf("Expected %d but got %d", nullTime.Time.Unix(), testNullTime.Unix())
+	}
+
+	testStrings := map[string]time.Time{
+		"2023-08-10 17:40:15.376736475-07:00": time.Unix(1691714415, 0),
+	}
+
+	for testString, expectedTime := range testStrings {
+		testJsonString := `{"testdate":"` + testString + `"}`
+		response := testJson{}
+
+		err := json.Unmarshal([]byte(testJsonString), &response)
+		if err != nil {
+			t.Error(err)
+		}
+
+		if response.Testdate.Time.Unix() != expectedTime.Unix() {
+			t.Errorf("Expected %d but got %d", expectedTime.Unix(), response.Testdate.Time.Unix())
+		}
+	}
+}

models/storageProvider.go

@@ -3,11 +3,12 @@ package models
 // StorageProvider is how a chunk storage provider should be implemented.
 type StorageProvider interface {
 	Setup() error
-	Save(filePath string, retryCount int) (string, error)
+	Save(localFilePath, destinationPath string, retryCount int) (string, error)
-
+	SetStreamId(streamID string)
-	SegmentWritten(localFilePath string)
+	SegmentWritten(localFilePath string) (string, int, error)
 	VariantPlaylistWritten(localFilePath string)
 	MasterPlaylistWritten(localFilePath string)
+	GetRemoteDestinationPathFromLocalFilePath(localFilePath string) string
 
 	Cleanup() error
 }

replays/clips.go

@@ -0,0 +1,141 @@
+package replays
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/db"
+	"github.com/owncast/owncast/utils"
+	"github.com/pkg/errors"
+	"github.com/teris-io/shortid"
+)
+
+// Clip represents a clip that has been created from a stream.
+// A clip is a subset of a stream that has has start and end seconds
+// relative to the start of the stream.
+type Clip struct {
+	ID                string    `json:"id"`
+	StreamId          string    `json:"stream_id"`
+	ClippedBy         string    `json:"clipped_by,omitempty"`
+	ClipTitle         string    `json:"title,omitempty"`
+	StreamTitle       string    `json:"stream_title,omitempty"`
+	RelativeStartTime float32   `json:"relativeStartTime"`
+	RelativeEndTime   float32   `json:"relativeEndTime"`
+	DurationSeconds   int       `json:"durationSeconds"`
+	Manifest          string    `json:"manifest,omitempty"`
+	Timestamp         time.Time `json:"timestamp"`
+}
+
+// GetClips will return all clips that have been recorded.
+func GetAllClips() ([]*Clip, error) {
+	clips, err := data.GetDatastore().GetQueries().GetAllClips(context.Background())
+	if err != nil {
+		return nil, errors.WithMessage(err, "failure to get clips")
+	}
+
+	response := []*Clip{}
+	for _, clip := range clips {
+		s := Clip{
+			ID:                clip.ID,
+			ClipTitle:         clip.ClipTitle.String,
+			StreamId:          clip.StreamID,
+			StreamTitle:       clip.StreamTitle.String,
+			RelativeStartTime: float32(clip.RelativeStartTime.Float64),
+			RelativeEndTime:   float32(clip.RelativeEndTime.Float64),
+			DurationSeconds:   int(clip.DurationSeconds),
+			Timestamp:         clip.Timestamp.Time,
+			Manifest:          fmt.Sprintf("/clip/%s", clip.ID),
+		}
+		response = append(response, &s)
+	}
+	return response, nil
+}
+
+// GetAllClipsForStream will return all clips that have been recorded for a stream.
+func GetAllClipsForStream(streamId string) ([]*Clip, error) {
+	clips, err := data.GetDatastore().GetQueries().GetAllClipsForStream(context.Background(), streamId)
+	if err != nil {
+		return nil, errors.WithMessage(err, "failure to get clips")
+	}
+
+	response := []*Clip{}
+	for _, clip := range clips {
+		s := Clip{
+			ID:                clip.ClipID,
+			ClipTitle:         clip.ClipTitle.String,
+			StreamTitle:       clip.StreamTitle.String,
+			RelativeStartTime: float32(clip.RelativeStartTime.Float64),
+			RelativeEndTime:   float32(clip.RelativeEndTime.Float64),
+			Timestamp:         clip.Timestamp.Time,
+			Manifest:          fmt.Sprintf("/clips/%s", clip.ClipID),
+		}
+		response = append(response, &s)
+	}
+	return response, nil
+}
+
+// AddClipForStream will save a new clip for a stream.
+func AddClipForStream(streamId, clipTitle, clippedBy string, relativeStartTimeSeconds, relativeEndTimeSeconds float32) (string, int, error) {
+	playlistGenerator := NewPlaylistGenerator()
+
+	// Verify this stream exists
+	if _, err := playlistGenerator.GetStream(streamId); err != nil {
+		return "", 0, errors.WithMessage(err, "stream not found")
+	}
+
+	// Verify this stream has at least one output configuration.
+	configs, err := playlistGenerator.GetConfigurationsForStream(streamId)
+	if err != nil {
+		return "", 0, errors.WithMessage(err, "unable to get configurations for stream")
+	}
+
+	if len(configs) == 0 {
+		return "", 0, errors.New("no configurations found for stream")
+	}
+
+	// We want the start and end seconds to be aligned to the segment so
+	// round up and down the values to get a fully inclusive segment range.
+	config := configs[0]
+	segmentDuration := int(config.SegmentDuration)
+
+	updatedRelativeStartTimeSeconds := utils.RoundDownToNearest(relativeStartTimeSeconds, segmentDuration)
+	updatedRelativeEndTimeSeconds := utils.RoundUpToNearest(relativeEndTimeSeconds, segmentDuration)
+	clipId := shortid.MustGenerate()
+	duration := updatedRelativeEndTimeSeconds - updatedRelativeStartTimeSeconds
+
+	err = data.GetDatastore().GetQueries().InsertClip(context.Background(), db.InsertClipParams{
+		ID:                clipId,
+		StreamID:          streamId,
+		ClipTitle:         sql.NullString{String: clipTitle, Valid: clipTitle != ""},
+		RelativeStartTime: sql.NullFloat64{Float64: float64(updatedRelativeStartTimeSeconds), Valid: true},
+		RelativeEndTime:   sql.NullFloat64{Float64: float64(updatedRelativeEndTimeSeconds), Valid: true},
+		Timestamp:         sql.NullTime{Time: time.Now(), Valid: true},
+	})
+	if err != nil {
+		return "", 0, errors.WithMessage(err, "failure to add clip")
+	}
+
+	return clipId, duration, nil
+}
+
+// GetFinalSegmentForStream will return the final known segment for a stream.
+func GetFinalSegmentForStream(streamId string) (*HLSSegment, error) {
+	segmentResponse, err := data.GetDatastore().GetQueries().GetFinalSegmentForStream(context.Background(), streamId)
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to get final segment for stream")
+	}
+
+	segment := HLSSegment{
+		ID:                    segmentResponse.ID,
+		StreamID:              segmentResponse.StreamID,
+		OutputConfigurationID: segmentResponse.OutputConfigurationID,
+		Path:                  segmentResponse.Path,
+		RelativeTimestamp:     segmentResponse.RelativeTimestamp,
+		Timestamp:             segmentResponse.Timestamp.Time,
+	}
+
+	return &segment, nil
+}

replays/hlsRecorder.go

@@ -0,0 +1,122 @@
+package replays
+
+import (
+	"context"
+	"database/sql"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/db"
+	"github.com/owncast/owncast/utils"
+	"github.com/teris-io/shortid"
+
+	log "github.com/sirupsen/logrus"
+)
+
+type HLSRecorder struct {
+	streamID  string
+	startTime time.Time
+
+	// The video variant configurations that were used for this stream.
+	outputConfigurations []HLSOutputConfiguration
+
+	datastore *data.Datastore
+}
+
+// NewRecording returns a new instance of the HLS recorder.
+func NewRecording(streamID string) *HLSRecorder {
+	// We don't support replaying offline clips.
+	if streamID == "offline" {
+		return nil
+	}
+
+	log.Infoln("Recording replay of this stream:", streamID)
+
+	h := HLSRecorder{
+		streamID:  streamID,
+		startTime: time.Now(),
+		datastore: data.GetDatastore(),
+	}
+
+	outputs := data.GetStreamOutputVariants()
+	latency := data.GetStreamLatencyLevel()
+
+	streamTitle := data.GetStreamTitle()
+	validTitle := streamTitle != ""
+
+	if err := h.datastore.GetQueries().InsertStream(context.Background(), db.InsertStreamParams{
+		ID:          streamID,
+		StartTime:   sql.NullTime{Time: h.startTime, Valid: true},
+		StreamTitle: sql.NullString{String: streamTitle, Valid: validTitle},
+	}); err != nil {
+		log.Panicln(err)
+	}
+
+	// Create a reference of the output configurations that were used for this stream.
+	for variantId, o := range outputs {
+		configId := shortid.MustGenerate()
+
+		if err := h.datastore.GetQueries().InsertOutputConfiguration(context.Background(), db.InsertOutputConfigurationParams{
+			ID:               configId,
+			Name:             o.Name,
+			StreamID:         streamID,
+			VariantID:        strconv.Itoa(variantId),
+			SegmentDuration:  int32(latency.SecondsPerSegment),
+			Bitrate:          int32(o.VideoBitrate),
+			Framerate:        int32(o.Framerate),
+			ResolutionWidth:  sql.NullInt32{Int32: int32(o.ScaledWidth), Valid: true},
+			ResolutionHeight: sql.NullInt32{Int32: int32(o.ScaledHeight), Valid: true},
+			Timestamp:        sql.NullTime{Time: time.Now(), Valid: true},
+		}); err != nil {
+			log.Panicln(err)
+		}
+
+		h.outputConfigurations = append(h.outputConfigurations, HLSOutputConfiguration{
+			ID:              configId,
+			Name:            o.Name,
+			VideoBitrate:    o.VideoBitrate,
+			ScaledWidth:     o.ScaledWidth,
+			ScaledHeight:    o.ScaledHeight,
+			Framerate:       o.Framerate,
+			SegmentDuration: float64(latency.SegmentCount),
+		})
+	}
+	return &h
+}
+
+// SegmentWritten is called when a segment is written to disk.
+func (h *HLSRecorder) SegmentWritten(path string) {
+	outputConfigurationIndexString := utils.GetIndexFromFilePath(path)
+	outputConfigurationIndex, err := strconv.Atoi(outputConfigurationIndexString)
+	if err != nil {
+		log.Errorln("HLSRecorder segmentWritten error:", err)
+		return
+	}
+
+	p := strings.ReplaceAll(path, "data/", "")
+	relativeTimestamp := time.Since(h.startTime)
+
+	if err := h.datastore.GetQueries().InsertSegment(context.Background(), db.InsertSegmentParams{
+		ID:                    shortid.MustGenerate(),
+		StreamID:              h.streamID,
+		OutputConfigurationID: h.outputConfigurations[outputConfigurationIndex].ID,
+		Path:                  p,
+		RelativeTimestamp:     float32(relativeTimestamp.Seconds()),
+		Timestamp:             sql.NullTime{Time: time.Now(), Valid: true},
+	}); err != nil {
+		log.Errorln(err)
+	}
+}
+
+// StreamEnded is called when a stream is ended so the end time can be noted
+// in the stream's metadata.
+func (h *HLSRecorder) StreamEnded() {
+	if err := h.datastore.GetQueries().SetStreamEnded(context.Background(), db.SetStreamEndedParams{
+		ID:      h.streamID,
+		EndTime: sql.NullTime{Time: time.Now(), Valid: true},
+	}); err != nil {
+		log.Errorln(err)
+	}
+}

replays/hlsSegment.go

@@ -0,0 +1,13 @@
+package replays
+
+import "time"
+
+// HLSSegment represents a single HLS segment.
+type HLSSegment struct {
+	ID                    string
+	StreamID              string
+	Timestamp             time.Time
+	RelativeTimestamp     float32
+	OutputConfigurationID string
+	Path                  string
+}

replays/mediaPlaylistAllowCacheTag.go

@@ -0,0 +1,49 @@
+package replays
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/grafov/m3u8"
+)
+
+// MediaPlaylistAllowCacheTag is a custom tag to explicitly state that this
+// playlist is allowed to be cached.
+type MediaPlaylistAllowCacheTag struct {
+	Type string
+}
+
+// TagName should return the full tag identifier including the leading
+// '#' and trailing ':' if the tag also contains a value or attribute
+// list.
+func (tag *MediaPlaylistAllowCacheTag) TagName() string {
+	return "#EXT-X-ALLOW-CACHE"
+}
+
+// Decode decodes the input line. The line will be the entire matched
+// line, including the identifier.
+func (tag *MediaPlaylistAllowCacheTag) Decode(line string) (m3u8.CustomTag, error) {
+	_, err := fmt.Sscanf(line, "#EXT-X-ALLOW-CACHE")
+
+	return tag, err
+}
+
+// SegmentTag specifies that this tag is not for segments.
+func (tag *MediaPlaylistAllowCacheTag) SegmentTag() bool {
+	return false
+}
+
+// Encode formats the structure to the text result.
+func (tag *MediaPlaylistAllowCacheTag) Encode() *bytes.Buffer {
+	buf := new(bytes.Buffer)
+
+	buf.WriteString(tag.TagName())
+	buf.WriteString(tag.Type)
+
+	return buf
+}
+
+// String implements Stringer interface.
+func (tag *MediaPlaylistAllowCacheTag) String() string {
+	return tag.Encode().String()
+}

replays/outputConfiguration.go

@@ -0,0 +1,27 @@
+package replays
+
+import "github.com/pkg/errors"
+
+type HLSOutputConfiguration struct {
+	ID              string
+	StreamId        string
+	VariantId       string
+	Name            string
+	VideoBitrate    int
+	ScaledWidth     int
+	ScaledHeight    int
+	Framerate       int
+	SegmentDuration float64
+}
+
+func (config *HLSOutputConfiguration) Validate() error {
+	if config.VideoBitrate == 0 {
+		return errors.New("video bitrate is unavailable")
+	}
+
+	if config.Framerate == 0 {
+		return errors.New("video framerate is unavailable")
+	}
+
+	return nil
+}

replays/playlistGenerator.go

@@ -0,0 +1,146 @@
+package replays
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/grafov/m3u8"
+	"github.com/pkg/errors"
+)
+
+// GetConfigurationsForStream returns the output configurations for a given stream.
+func (p *PlaylistGenerator) GetConfigurationsForStream(streamId string) ([]*HLSOutputConfiguration, error) {
+	outputConfigRows, err := p.datastore.GetQueries().GetOutputConfigurationsForStreamId(context.Background(), streamId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get output configurations for stream")
+	}
+
+	outputConfigs := []*HLSOutputConfiguration{}
+	for _, row := range outputConfigRows {
+		config := &HLSOutputConfiguration{
+			ID:              row.ID,
+			StreamId:        streamId,
+			VariantId:       row.VariantID,
+			Name:            row.Name,
+			VideoBitrate:    int(row.Bitrate),
+			Framerate:       int(row.Framerate),
+			ScaledHeight:    int(row.ResolutionWidth.Int32),
+			ScaledWidth:     int(row.ResolutionHeight.Int32),
+			SegmentDuration: float64(row.SegmentDuration),
+		}
+		outputConfigs = append(outputConfigs, config)
+	}
+
+	return outputConfigs, nil
+}
+
+func (p *PlaylistGenerator) createMediaPlaylistForConfigurationAndSegments(configuration *HLSOutputConfiguration, startTime time.Time, inProgress bool, segments []HLSSegment) (*m3u8.MediaPlaylist, error) {
+	playlistSize := len(segments)
+	segmentDuration := configuration.SegmentDuration
+	playlist, err := m3u8.NewMediaPlaylist(0, uint(playlistSize))
+
+	playlist.TargetDuration = configuration.SegmentDuration
+
+	if !inProgress {
+		playlist.MediaType = m3u8.VOD
+	} else {
+		playlist.MediaType = m3u8.EVENT
+	}
+
+	// Add the segments to the playlist.
+	for index, segment := range segments {
+		// If it's a URL leave it as is, if it's a local path then append a slash.
+		path := segment.Path
+		if !strings.HasPrefix(path, "http") {
+			path = "/" + path
+		}
+
+		mediaSegment := m3u8.MediaSegment{
+			URI:             path,
+			Duration:        segmentDuration,
+			SeqId:           uint64(index),
+			ProgramDateTime: segment.Timestamp,
+		}
+		if err := playlist.AppendSegment(&mediaSegment); err != nil {
+			return nil, errors.Wrap(err, "failed to append segment to recording playlist")
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	// Configure the properties of this media playlist.
+	if err := playlist.SetProgramDateTime(startTime); err != nil {
+		return nil, errors.Wrap(err, "failed to set media playlist program date time")
+	}
+
+	// Our live output is specified as v6, so let's match it to be as close as
+	// possible to what we're doing for live streams.
+	playlist.SetVersion(6)
+
+	if !inProgress {
+		// Specify explicitly that the playlist content is allowed to be cached.
+		// However, if in-progress recordings are supported this should not be enabled
+		// in order for the playlist to be updated with new segments. inProgress is
+		// determined by seeing if the stream has an endTime or not.
+		playlist.SetCustomTag(&MediaPlaylistAllowCacheTag{})
+
+		// Set the ENDLIST tag and close the playlist for writing if the stream is
+		// not still in progress.
+		playlist.Close()
+	}
+
+	return playlist, nil
+}
+
+func (p *PlaylistGenerator) createNewMasterPlaylist() *m3u8.MasterPlaylist {
+	playlist := m3u8.NewMasterPlaylist()
+	playlist.SetIndependentSegments(true)
+	playlist.SetVersion(6)
+
+	return playlist
+}
+
+// GetAllSegmentsForOutputConfiguration returns all the segments for a given output config.
+func (p *PlaylistGenerator) GetAllSegmentsForOutputConfiguration(outputId string) ([]HLSSegment, error) {
+	segmentRows, err := p.datastore.GetQueries().GetSegmentsForOutputId(context.Background(), outputId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get segments for output config")
+	}
+
+	segments := []HLSSegment{}
+	for _, row := range segmentRows {
+		segment := HLSSegment{
+			ID:                    row.ID,
+			StreamID:              row.StreamID,
+			OutputConfigurationID: row.OutputConfigurationID,
+			Timestamp:             row.Timestamp.Time,
+			Path:                  row.Path,
+		}
+		segments = append(segments, segment)
+	}
+
+	return segments, nil
+}
+
+func (p *PlaylistGenerator) getMediaPlaylistParamsForConfig(config *HLSOutputConfiguration) m3u8.VariantParams {
+	params := m3u8.VariantParams{
+		ProgramId: 1,
+		Name:      config.Name,
+		FrameRate: float64(config.Framerate),
+		Bandwidth: uint32(config.VideoBitrate * 1000),
+		// Match what is generated in our live playlists.
+		Codecs: "avc1.64001f,mp4a.40.2",
+	}
+
+	// If both the width and height are set then we can set that as
+	// the resolution in the media playlist.
+	if config.ScaledHeight > 0 && config.ScaledWidth > 0 {
+		params.Resolution = fmt.Sprintf("%dx%d", config.ScaledWidth, config.ScaledHeight)
+	}
+
+	return params
+}

replays/playlistGenerator_test.go

@@ -0,0 +1,136 @@
+package replays
+
+import (
+	"testing"
+	"time"
+
+	"github.com/grafov/m3u8"
+)
+
+var (
+	generator = NewPlaylistGenerator()
+	config    = []HLSOutputConfiguration{
+		{
+			ID:           "1",
+			VideoBitrate: 1000,
+			Framerate:    30,
+		},
+		{
+			ID:           "2",
+			VideoBitrate: 2000,
+			Framerate:    30,
+		},
+	}
+)
+
+var segments = []HLSSegment{
+	{
+		ID:                    "testSegmentId",
+		StreamID:              "testStreamId",
+		Timestamp:             time.Now(),
+		OutputConfigurationID: "testOutputConfigId",
+		Path:                  "hls/testStreamId/testOutputConfigId/testSegmentId.ts",
+	},
+}
+
+func TestMasterPlaylist(t *testing.T) {
+	playlist := generator.createNewMasterPlaylist()
+
+	mediaPlaylists, err := generator.createMediaPlaylistForConfigurationAndSegments(&config[0], time.Now(), false, segments)
+	playlist.Append("test", mediaPlaylists, m3u8.VariantParams{
+		Bandwidth: uint32(config[0].VideoBitrate),
+		FrameRate: float64(config[0].Framerate),
+	})
+	mediaPlaylists.Close()
+
+	if err != nil {
+		t.Error(err)
+	}
+
+	if playlist.Version() != 6 {
+		t.Error("expected version 6, got", playlist.Version())
+	}
+
+	if !playlist.IndependentSegments() {
+		t.Error("expected independent segments")
+	}
+
+	if playlist.Variants[0].Bandwidth != uint32(config[0].VideoBitrate) {
+		t.Error("expected bandwidth", config[0].VideoBitrate, "got", playlist.Variants[0].Bandwidth)
+	}
+
+	if playlist.Variants[0].FrameRate != float64(config[0].Framerate) {
+		t.Error("expected framerate", config[0].Framerate, "got", playlist.Variants[0].FrameRate)
+	}
+}
+
+func TestCompletedMediaPlaylist(t *testing.T) {
+	startTime := segments[0].Timestamp
+	conf := config[0]
+
+	// Create a completed media playlist.
+	playlist, err := generator.createMediaPlaylistForConfigurationAndSegments(&conf, startTime, false, segments)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if playlist.TargetDuration != conf.SegmentDuration {
+		t.Error("expected target duration", conf.SegmentDuration, "got", playlist.TargetDuration)
+	}
+
+	// Verify it's marked as cachable.
+	if playlist.Custom["#EXT-X-ALLOW-CACHE"].String() != "#EXT-X-ALLOW-CACHE" {
+		t.Error("expected cachable playlist, tag not set")
+	}
+
+	// Verify it has the correct number of segments in the media playlist.
+	if int(playlist.Count()) != len(segments) {
+		t.Error("expected", len(segments), "segments, got", playlist.Count())
+	}
+
+	// Test the playlist version.
+	if playlist.Version() != 6 {
+		t.Error("expected version 6, got", playlist.Version())
+	}
+
+	// Verify the playlist type
+	if playlist.MediaType != m3u8.VOD {
+		t.Error("expected VOD playlist type, got type", playlist.MediaType)
+	}
+
+	// Verify the first segment URI.
+	if playlist.Segments[0].URI != "/"+segments[0].Path {
+		t.Error("expected segment URI", segments[0].Path, "got", playlist.Segments[0].URI)
+	}
+}
+
+func TestInProgressMediaPlaylist(t *testing.T) {
+	startTime := segments[0].Timestamp
+	conf := config[0]
+
+	// Create a completed media playlist.
+	playlist, err := generator.createMediaPlaylistForConfigurationAndSegments(&conf, startTime, true, segments)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// Verify it's marked as cachable.
+	if playlist.Custom != nil && playlist.Custom["#EXT-X-ALLOW-CACHE"].String() == "#EXT-X-ALLOW-CACHE" {
+		t.Error("expected non-achable playlist when stream is still in progress")
+	}
+
+	// Verify it has the correct number of segments in the media playlist.
+	if int(playlist.Count()) != len(segments) {
+		t.Error("expected", len(segments), "segments, got", playlist.Count())
+	}
+
+	// Test the playlist version.
+	if playlist.Version() != 6 {
+		t.Error("expected version 6, got", playlist.Version())
+	}
+
+	// Verify the playlist type
+	if playlist.MediaType != m3u8.EVENT {
+		t.Error("expected EVENT playlist type, got type", playlist.MediaType)
+	}
+}

replays/replay_test.go

@@ -0,0 +1,154 @@
+package replays
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/db"
+	"github.com/owncast/owncast/utils"
+	log "github.com/sirupsen/logrus"
+	"github.com/teris-io/shortid"
+)
+
+var (
+	fakeStreamId        = shortid.MustGenerate()
+	fakeSegmentCount    = 300
+	fakeSegmentDuration = 0
+	fakeStreamStartTime = time.Now()
+	fakeConfigId        = ""
+	fakeClipper         = shortid.MustGenerate()
+	fakeClipStartTime   = 10
+	fakeClipEndTime     = 15
+)
+
+func TestMain(m *testing.M) {
+	if err := data.SetupPersistence(":memory:"); err != nil {
+		panic(err)
+	}
+
+	Setup()
+	populateFakeStream()
+	m.Run()
+}
+
+func populateFakeStream() {
+	queries := data.GetDatastore().GetQueries()
+
+	recording := NewRecording(fakeStreamId)
+	fakeConfigId = recording.outputConfigurations[0].ID
+	fakeSegmentDuration = data.GetStreamLatencyLevel().SecondsPerSegment // Seconds
+
+	for i := 0; i < fakeSegmentCount; i++ {
+		fakeSegmentName := fmt.Sprintf("%s-%d.ts", fakeStreamId, i)
+		if err := queries.InsertSegment(context.Background(), db.InsertSegmentParams{
+			ID:                    shortid.MustGenerate(),
+			StreamID:              fakeStreamId,
+			OutputConfigurationID: fakeConfigId,
+			Path:                  filepath.Join(fakeStreamId, fakeConfigId, "0", fakeSegmentName),
+			RelativeTimestamp:     float32(i * fakeSegmentDuration),
+			Timestamp:             sql.NullTime{Time: fakeStreamStartTime.Add(time.Duration(fakeSegmentDuration * i)), Valid: true},
+		}); err != nil {
+			log.Errorln(err)
+		}
+	}
+
+	if err := queries.SetStreamEnded(context.Background(), db.SetStreamEndedParams{
+		ID:      fakeStreamId,
+		EndTime: sql.NullTime{Time: fakeStreamStartTime.Add(time.Duration(fakeSegmentDuration * fakeSegmentCount)), Valid: true},
+	}); err != nil {
+		log.Errorln(err)
+	}
+}
+
+func TestStream(t *testing.T) {
+	playlist := NewPlaylistGenerator()
+	stream, err := playlist.GetStream(fakeStreamId)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if stream.ID != fakeStreamId {
+		t.Error("expected stream id", fakeStreamId, "got", stream.ID)
+	}
+}
+
+func TestPlaylist(t *testing.T) {
+	playlist := NewPlaylistGenerator()
+	p, err := playlist.GenerateMediaPlaylistForStreamAndConfiguration(fakeStreamId, fakeConfigId)
+	if p == nil {
+		t.Error("expected playlist")
+	}
+
+	if err != nil {
+		t.Error(err)
+	}
+
+	if len(p.Segments) != fakeSegmentCount {
+		t.Error("expected", fakeSegmentCount, "segments, got", len(p.Segments))
+	}
+}
+
+func TestClip(t *testing.T) {
+	segmentDuration := data.GetStreamLatencyLevel().SecondsPerSegment
+	playlist := NewPlaylistGenerator()
+	clipId, _, err := AddClipForStream(fakeStreamId, "test clip", fakeClipper, float32(fakeClipStartTime), float32(fakeClipEndTime))
+	if err != nil {
+		t.Error(err)
+	}
+
+	clips, err := GetAllClips()
+	if err != nil {
+		t.Error(err)
+	}
+
+	if len(clips) != 1 {
+		t.Error("expected 1 clip, got", len(clips))
+	}
+
+	clip := clips[0]
+
+	if clip.ID != clipId {
+		t.Error("expected clip id", clipId, "got", clip.ID)
+	}
+
+	if clip.Manifest != fmt.Sprintf("/clip/%s", clipId) {
+		t.Error("expected manifest id", fmt.Sprintf("/clip/%s", clipId), "got", clip.Manifest)
+	}
+
+	expectedStartTime := float32(utils.RoundDownToNearest(float32(fakeClipStartTime), segmentDuration))
+	if clip.RelativeStartTime != expectedStartTime {
+		t.Error("expected clip start time", fakeClipStartTime, "got", clip.RelativeStartTime)
+	}
+
+	expectedEndTime := float32(utils.RoundUpToNearest(float32(fakeClipEndTime), segmentDuration))
+	if clip.RelativeEndTime != expectedEndTime {
+		t.Error("expected clip end time", fakeClipEndTime, "got", clip.RelativeEndTime)
+	}
+
+	expectedDuration := expectedEndTime - expectedStartTime
+	if float32(clip.DurationSeconds) != expectedDuration {
+		t.Error("expected clip duration", expectedDuration, "got", clip.DurationSeconds)
+	}
+
+	p, err := playlist.GenerateMediaPlaylistForClipAndConfiguration(clipId, fakeConfigId)
+	if err != nil {
+		t.Error(err)
+	}
+	if p == nil {
+		t.Error("expected playlist")
+	}
+
+	expectedSegmentCount := 3
+	if len(p.Segments) != expectedSegmentCount {
+		t.Error("expected", expectedSegmentCount, "segments, got", len(p.Segments))
+	}
+
+	if p.TargetDuration != float64(fakeSegmentDuration) {
+		t.Error("expected target duration of", fakeSegmentDuration, "got", p.TargetDuration)
+	}
+}

replays/replays.go

@@ -0,0 +1,21 @@
+package replays
+
+import (
+	"context"
+
+	"github.com/owncast/owncast/core/data"
+	log "github.com/sirupsen/logrus"
+)
+
+// Setup will setup the replay package.
+func Setup() {
+	fixUnfinishedStreams()
+}
+
+// fixUnfinishedStreams will find streams with no end time and attempt to
+// give them an end time based on the last segment assigned to that stream.
+func fixUnfinishedStreams() {
+	if err := data.GetDatastore().GetQueries().FixUnfinishedStreams(context.Background()); err != nil {
+		log.Warnln(err)
+	}
+}

replays/storageProvider.go

@@ -0,0 +1,6 @@
+package replays
+
+type StorageProvider interface {
+	Setup() error
+	Save(localFilePath, destinationPath string, retryCount int) (string, error)
+}

replays/stream.go

@@ -0,0 +1,41 @@
+package replays
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/owncast/owncast/core/data"
+	"github.com/pkg/errors"
+)
+
+type Stream struct {
+	ID         string    `json:"id"`
+	Title      string    `json:"title,omitempty"`
+	StartTime  time.Time `json:"startTime"`
+	EndTime    time.Time `json:"endTime,omitempty"`
+	InProgress bool      `json:"inProgress,omitempty"`
+	Manifest   string    `json:"manifest,omitempty"`
+}
+
+// GetStreams will return all streams that have been recorded.
+func GetStreams() ([]*Stream, error) {
+	streams, err := data.GetDatastore().GetQueries().GetStreams(context.Background())
+	if err != nil {
+		return nil, errors.WithMessage(err, "failure to get streams")
+	}
+
+	response := []*Stream{}
+	for _, stream := range streams {
+		s := Stream{
+			ID:         stream.ID,
+			Title:      stream.StreamTitle.String,
+			StartTime:  stream.StartTime.Time,
+			EndTime:    stream.EndTime.Time,
+			InProgress: !stream.EndTime.Valid,
+			Manifest:   fmt.Sprintf("/replay/%s", stream.ID),
+		}
+		response = append(response, &s)
+	}
+	return response, nil
+}

replays/streamClipPlaylistGenerator.go

@@ -0,0 +1,142 @@
+package replays
+
+import (
+	"context"
+	"strings"
+
+	"github.com/grafov/m3u8"
+	"github.com/owncast/owncast/db"
+	"github.com/owncast/owncast/models"
+	"github.com/pkg/errors"
+)
+
+// GenerateMasterPlaylistForClip returns a master playlist for a given clip Id.
+// It includes references to the media playlists for each output configuration.
+func (p *PlaylistGenerator) GenerateMasterPlaylistForClip(clipId string) (*m3u8.MasterPlaylist, error) {
+	clip, err := p.datastore.GetQueries().GetClip(context.Background(), clipId)
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to fetch requested clip")
+	}
+
+	streamId := clip.StreamID
+	configs, err := p.GetConfigurationsForStream(streamId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get configurations for stream")
+	}
+
+	// Create the master playlist that will hold the different media playlists.
+	masterPlaylist := p.createNewMasterPlaylist()
+
+	// Create the media playlists for each output configuration.
+	for _, config := range configs {
+		// Verify the validity of the configuration.
+		if err := config.Validate(); err != nil {
+			return nil, errors.Wrap(err, "invalid output configuration")
+		}
+
+		mediaPlaylist, err := p.GenerateMediaPlaylistForClipAndConfiguration(clipId, config.ID)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to create clip media playlist")
+		}
+
+		// Append the media playlist to the master playlist.
+		params := p.getMediaPlaylistParamsForConfig(config)
+
+		// Add the media playlist to the master playlist.
+		publicPlaylistPath := strings.Join([]string{"/clip", clipId, config.ID}, "/")
+		masterPlaylist.Append(publicPlaylistPath, mediaPlaylist, params)
+	}
+
+	// Return the final master playlist that contains all the media playlists.
+	return masterPlaylist, nil
+}
+
+// GenerateMediaPlaylistForClipAndConfiguration returns a media playlist for a
+// given clip Id and output configuration.
+func (p *PlaylistGenerator) GenerateMediaPlaylistForClipAndConfiguration(clipId, outputConfigurationId string) (*m3u8.MediaPlaylist, error) {
+	clip, err := p.GetClip(clipId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get stream")
+	}
+
+	config, err := p.GetOutputConfig(outputConfigurationId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get output configuration")
+	}
+
+	clipStartSeconds := clip.RelativeStartTime
+	clipEndSeconds := clip.RelativeEndTime
+
+	// Fetch all the segments for this configuration.
+	segments, err := p.GetAllSegmentsForOutputConfigurationAndWindow(outputConfigurationId, clipStartSeconds, clipEndSeconds)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get all clip segments for output configuration")
+	}
+
+	// Create the media playlist for this configuration and add the segments.
+	mediaPlaylist, err := p.createMediaPlaylistForConfigurationAndSegments(config, clip.Timestamp, false, segments)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create clip media playlist")
+	}
+
+	return mediaPlaylist, nil
+}
+
+// GetClip returns a clip by its ID.
+func (p *PlaylistGenerator) GetClip(clipId string) (*Clip, error) {
+	clip, err := p.datastore.GetQueries().GetClip(context.Background(), clipId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get clip")
+	}
+
+	if clip.ClipID == "" {
+		return nil, errors.Wrap(err, "failed to get clip")
+	}
+
+	if !clip.RelativeEndTime.Valid {
+		return nil, errors.Wrap(err, "failed to get clip")
+	}
+
+	timestamp, err := models.FlexibleDateParse(clip.ClipTimestamp)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to parse clip timestamp")
+	}
+
+	c := Clip{
+		ID:                clip.ClipID,
+		StreamId:          clip.StreamID,
+		ClipTitle:         clip.ClipTitle.String,
+		RelativeStartTime: float32(clip.RelativeStartTime.Float64),
+		RelativeEndTime:   float32(clip.RelativeEndTime.Float64),
+		Timestamp:         timestamp,
+	}
+
+	return &c, nil
+}
+
+// GetAllSegmentsForOutputConfigurationAndWindow returns all the segments for a
+// given output config and time window.
+func (p *PlaylistGenerator) GetAllSegmentsForOutputConfigurationAndWindow(configId string, startSeconds, endSeconds float32) ([]HLSSegment, error) {
+	segmentRows, err := p.datastore.GetQueries().GetSegmentsForOutputIdAndWindow(context.Background(), db.GetSegmentsForOutputIdAndWindowParams{
+		OutputConfigurationID: configId,
+		StartSeconds:          startSeconds,
+		EndSeconds:            endSeconds,
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get clip segments for output config")
+	}
+
+	segments := []HLSSegment{}
+	for _, row := range segmentRows {
+		segment := HLSSegment{
+			ID:                    row.ID,
+			StreamID:              row.StreamID,
+			OutputConfigurationID: row.OutputConfigurationID,
+			Timestamp:             row.Timestamp.Time,
+			Path:                  row.Path,
+		}
+		segments = append(segments, segment)
+	}
+
+	return segments, nil
+}

replays/streamReplayPlaylistGenerator.go

@@ -0,0 +1,129 @@
+package replays
+
+import (
+	"context"
+	"strings"
+
+	"github.com/grafov/m3u8"
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/db"
+	"github.com/pkg/errors"
+)
+
+/*
+The PlaylistGenerator is responsible for creating the master and media
+playlists, in order to replay a stream in whole, or part. It requires detailed
+metadata about how the initial live stream was configured, as well as a
+access to every segment that was created during the live stream.
+*/
+
+type PlaylistGenerator struct {
+	datastore *data.Datastore
+}
+
+func NewPlaylistGenerator() *PlaylistGenerator {
+	return &PlaylistGenerator{
+		datastore: data.GetDatastore(),
+	}
+}
+
+func (p *PlaylistGenerator) GenerateMasterPlaylistForStream(streamId string) (*m3u8.MasterPlaylist, error) {
+	// Determine the different output configurations for this stream.
+	configs, err := p.GetConfigurationsForStream(streamId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get configurations for stream")
+	}
+
+	// Create the master playlist that will hold the different media playlists.
+	masterPlaylist := p.createNewMasterPlaylist()
+
+	// Create the media playlists for each output configuration.
+	for _, config := range configs {
+		// Verify the validity of the configuration.
+		if err := config.Validate(); err != nil {
+			return nil, errors.Wrap(err, "invalid output configuration")
+		}
+
+		mediaPlaylist, err := p.GenerateMediaPlaylistForStreamAndConfiguration(streamId, config.ID)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to create media playlist")
+		}
+
+		// Append the media playlist to the master playlist.
+		params := p.getMediaPlaylistParamsForConfig(config)
+
+		// Add the media playlist to the master playlist.
+		publicPlaylistPath := strings.Join([]string{"/replay", streamId, config.ID}, "/")
+		masterPlaylist.Append(publicPlaylistPath, mediaPlaylist, params)
+	}
+
+	// Return the final master playlist that contains all the media playlists.
+	return masterPlaylist, nil
+}
+
+func (p *PlaylistGenerator) GenerateMediaPlaylistForStreamAndConfiguration(streamId, outputConfigurationId string) (*m3u8.MediaPlaylist, error) {
+	stream, err := p.GetStream(streamId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get stream")
+	}
+
+	config, err := p.GetOutputConfig(outputConfigurationId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get output configuration")
+	}
+
+	// Fetch all the segments for this configuration.
+	segments, err := p.GetAllSegmentsForOutputConfiguration(outputConfigurationId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get all segments for output configuration")
+	}
+
+	// Create the media playlist for this configuration and add the segments.
+	mediaPlaylist, err := p.createMediaPlaylistForConfigurationAndSegments(config, stream.StartTime, stream.InProgress, segments)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create media playlist")
+	}
+
+	return mediaPlaylist, nil
+}
+
+func (p *PlaylistGenerator) GetStream(streamId string) (*Stream, error) {
+	stream, err := p.datastore.GetQueries().GetStreamById(context.Background(), streamId)
+	if stream.ID == "" {
+		return nil, errors.Wrap(err, "failed to get stream")
+	}
+
+	s := Stream{
+		ID:         stream.ID,
+		Title:      stream.StreamTitle.String,
+		StartTime:  stream.StartTime.Time,
+		EndTime:    stream.EndTime.Time,
+		InProgress: !stream.EndTime.Valid,
+	}
+
+	return &s, nil
+}
+
+func (p *PlaylistGenerator) GetOutputConfig(outputConfigId string) (*HLSOutputConfiguration, error) {
+	config, err := p.datastore.GetQueries().GetOutputConfigurationForId(context.Background(), outputConfigId)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get output configuration")
+	}
+
+	return createConfigFromConfigRow(config), nil
+}
+
+func createConfigFromConfigRow(row db.GetOutputConfigurationForIdRow) *HLSOutputConfiguration {
+	config := HLSOutputConfiguration{
+		ID:              row.ID,
+		StreamId:        row.StreamID,
+		VariantId:       row.VariantID,
+		Name:            row.Name,
+		VideoBitrate:    int(row.Bitrate),
+		Framerate:       int(row.Framerate),
+		ScaledHeight:    int(row.ResolutionWidth.Int32),
+		ScaledWidth:     int(row.ResolutionHeight.Int32),
+		SegmentDuration: float64(row.SegmentDuration),
+	}
+	return &config
+}

router/router.go

@@ -392,6 +392,15 @@ func Start() error {
 	http.HandleFunc("/api/auth/fediverse", middleware.RequireUserAccessToken(fediverseauth.RegisterFediverseOTPRequest))
 	http.HandleFunc("/api/auth/fediverse/verify", fediverseauth.VerifyFediverseOTPRequest)
 
+	// Replay functionality. This route handles both /replay/{streamId} (master)
+	// and /replay/{streamId}/{outputConfigId} (media) routes.
+	http.HandleFunc("/api/replays", controllers.GetReplays)
+	http.HandleFunc("/replay/", controllers.GetReplay)
+
+	http.HandleFunc("/api/clips", controllers.GetAllClips)
+	http.HandleFunc("/api/clip", controllers.AddClip)
+	http.HandleFunc("/clip/", controllers.GetClip)
+
 	// ActivityPub has its own router
 	activitypub.Start(data.GetDatastore())
 

static/web/_next/static/QNBMvcxExtLzshYBDiHgr/_ssgManifest.js

@@ -0,0 +1 @@
+self.__SSG_MANIFEST=new Set,self.__SSG_MANIFEST_CB&&self.__SSG_MANIFEST_CB();

test/automated/replays/package.json

@@ -0,0 +1,18 @@
+{
+  "name": "owncast-test-automation",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "jest --bail"
+  },
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "m3u8-parser": "^4.7.0",
+    "node-fetch": "^2.6.7"
+  },
+  "devDependencies": {
+    "jest": "^26.6.3"
+  }
+}

test/automated/replays/replays.test.js

@@ -0,0 +1,130 @@
+const m3u8Parser = require('m3u8-parser');
+const fetch = require('node-fetch');
+const url = require('url');
+const { test } = require('@jest/globals');
+
+const REPLAYS_API = '/api/replays';
+const TEST_OWNCAST_INSTANCE = 'http://localhost:8080';
+const HLS_FETCH_ITERATIONS = 5;
+
+jest.setTimeout(40000);
+
+async function getPlaylist(urlString) {
+	const response = await fetch(urlString);
+	expect(response.status).toBe(200);
+	const body = await response.text();
+
+	var parser = new m3u8Parser.Parser();
+
+	parser.push(body);
+	parser.end();
+
+	return parser.manifest;
+}
+
+async function getReplaysAPI(urlString) {
+	const response = await fetch(urlString);
+	expect(response.status).toBe(200);
+	const body = await response.text();
+	return body;
+}
+
+function normalizeUrl(urlString, baseUrl) {
+	let parsedString = url.parse(urlString);
+	if (!parsedString.host) {
+		const testInstanceRoot = url.parse(baseUrl);
+		parsedString.protocol = testInstanceRoot.protocol;
+		parsedString.host = testInstanceRoot.host;
+
+		const filename = baseUrl.substring(baseUrl.lastIndexOf('/') + 1);
+		parsedString.pathname =
+			testInstanceRoot.pathname.replace(filename, '') + urlString;
+	}
+	return url.format(parsedString).toString();
+}
+
+// Iterate over an array of video segments and make sure they return back
+// valid status.
+async function validateSegments(segments) {
+	for (let segment of segments) {
+		const res = await fetch(segment);
+		expect(res.status).toBe(200);
+	}
+}
+
+describe('fetch list of clips', () => {
+	const replaysAPIEndpoint = `${TEST_OWNCAST_INSTANCE}${REPLAYS_API}`;
+	// var masterPlaylist;
+	// var mediaPlaylistUrl;
+
+	test('fetch replay list', async (done) => {
+		console.log(replaysAPIEndpoint);
+		try {
+			const response = await getReplaysAPI(replaysAPIEndpoint);
+			console.log(response);
+		} catch (e) {
+			console.error('error fetching and parsing master playlist', e);
+		}
+
+		done();
+	});
+
+	// test('verify there is a media playlist', () => {
+	// 	// Master playlist should have at least one media playlist.
+	// 	expect(masterPlaylist.playlists.length).toBe(1);
+
+	// 	try {
+	// 		mediaPlaylistUrl = normalizeUrl(
+	// 			masterPlaylist.playlists[0].uri,
+	// 			masterPlaylistUrl
+	// 		);
+	// 	} catch (e) {
+	// 		console.error('error fetching and parsing media playlist', e);
+	// 	}
+	// });
+
+	// test('verify there are segments', async (done) => {
+	// 	let playlist;
+	// 	try {
+	// 		playlist = await getPlaylist(mediaPlaylistUrl);
+	// 	} catch (e) {
+	// 		console.error('error verifying segments in media playlist', e);
+	// 	}
+
+	// 	const segments = playlist.segments;
+	// 	expect(segments.length).toBeGreaterThan(0);
+
+	// 	done();
+	// });
+
+	// // Iterate over segments and make sure they change.
+	// // Use the reported duration of the segment to wait to
+	// // fetch another just like a real HLS player would do.
+	// var lastSegmentUrl;
+	// for (let i = 0; i < HLS_FETCH_ITERATIONS; i++) {
+	// 	test('fetch and monitor media playlist segments ' + i, async (done) => {
+	// 		await new Promise((r) => setTimeout(r, 5000));
+
+	// 		try {
+	// 			var playlist = await getPlaylist(mediaPlaylistUrl);
+	// 		} catch (e) {
+	// 			console.error('error updating media playlist', mediaPlaylistUrl, e);
+	// 		}
+
+	// 		const segments = playlist.segments;
+	// 		const segment = segments[segments.length - 1];
+	// 		expect(segment.uri).not.toBe(lastSegmentUrl);
+
+	// 		try {
+	// 			var segmentUrl = normalizeUrl(segment.uri, mediaPlaylistUrl);
+	// 			await validateSegments([segmentUrl]);
+	// 		} catch (e) {
+	// 			console.error('unable to validate HLS segment', segmentUrl, e);
+	// 		}
+
+	// 		lastSegmentUrl = segment.uri;
+
+	// 		done();
+	// 	});
+	// }
+});

test/automated/replays/run.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+
+set -e
+
+source ../tools.sh
+
+# Install the node test framework
+npm install --silent >/dev/null
+
+install_ffmpeg
+
+start_owncast "--enableReplayFeatures"
+
+start_stream
+
+sleep 10
+
+# Run tests against a fresh install with no settings.
+npm test

test/automated/tools.sh

@@ -3,91 +3,96 @@
 set -e
 
 function install_ffmpeg() {
-    # install a specific version of ffmpeg
+	# install a specific version of ffmpeg
 
-    FFMPEG_VER="4.4.1"
+	FFMPEG_VER="4.4.1"
-    FFMPEG_PATH="$(pwd)/ffmpeg-$FFMPEG_VER"
+	FFMPEG_PATH="$(pwd)/ffmpeg-$FFMPEG_VER"
-    PATH=$FFMPEG_PATH:$PATH
+	PATH=$FFMPEG_PATH:$PATH
 
-    if ! [[ -d "$FFMPEG_PATH" ]]; then
+	if ! [[ -d "$FFMPEG_PATH" ]]; then
-        mkdir "$FFMPEG_PATH"
+		mkdir "$FFMPEG_PATH"
-    fi
+	fi
 
-    pushd "$FFMPEG_PATH" >/dev/null
+	pushd "$FFMPEG_PATH" >/dev/null
 
-    if [[ -x "$FFMPEG_PATH/ffmpeg" ]]; then
+	if [[ -x "$FFMPEG_PATH/ffmpeg" ]]; then
-    
-        ffmpeg_version=$("$FFMPEG_PATH/ffmpeg" -version | awk -F 'ffmpeg version' '{print $2}' | awk 'NR==1{print $1}')
 
-        if [[ "$ffmpeg_version" == "$FFMPEG_VER-static" ]]; then
+		ffmpeg_version=$("$FFMPEG_PATH/ffmpeg" -version | awk -F 'ffmpeg version' '{print $2}' | awk 'NR==1{print $1}')
-            popd >/dev/null
-            return 0
-        else
-            mv "$FFMPEG_PATH/ffmpeg" "$FFMPEG_PATH/ffmpeg.bk" || rm -f "$FFMPEG_PATH/ffmpeg"
-        fi
-    fi
 
-    rm -f ffmpeg.zip
+		if [[ "$ffmpeg_version" == "$FFMPEG_VER-static" ]]; then
-    curl -sL --fail https://github.com/ffbinaries/ffbinaries-prebuilt/releases/download/v${FFMPEG_VER}/ffmpeg-${FFMPEG_VER}-linux-64.zip --output ffmpeg.zip >/dev/null
+			popd >/dev/null
-    unzip -o ffmpeg.zip >/dev/null && rm -f ffmpeg.zip
+			return 0
-    chmod +x ffmpeg
+		else
-    PATH=$FFMPEG_PATH:$PATH
+			mv "$FFMPEG_PATH/ffmpeg" "$FFMPEG_PATH/ffmpeg.bk" || rm -f "$FFMPEG_PATH/ffmpeg"
+		fi
+	fi
 
-    popd >/dev/null
+	rm -f ffmpeg.zip
+	curl -sL --fail https://github.com/ffbinaries/ffbinaries-prebuilt/releases/download/v${FFMPEG_VER}/ffmpeg-${FFMPEG_VER}-linux-64.zip --output ffmpeg.zip >/dev/null
+	unzip -o ffmpeg.zip >/dev/null && rm -f ffmpeg.zip
+	chmod +x ffmpeg
+	PATH=$FFMPEG_PATH:$PATH
+
+	popd >/dev/null
 }
 
 function start_owncast() {
-    # Build and run owncast from source
+	# Build and run owncast from source
-    echo "Building owncast..."
+	echo "Building owncast..."
-    pushd "$(git rev-parse --show-toplevel)" >/dev/null
+	pushd "$(git rev-parse --show-toplevel)" >/dev/null
-    go build -o owncast main.go
+	go build -o owncast main.go
+
+	if [ -z "$1" ]; then
+		echo "Running owncast..."
+	else
+		echo "Running owncast with flags: $1"
+	fi
 
-    echo "Running owncast..."
+	./owncast -database "$TEMP_DB" $1 &
-    ./owncast -database "$TEMP_DB" &
+	SERVER_PID=$!
-    SERVER_PID=$!
+	popd >/dev/null
-    popd >/dev/null
 
-    sleep 5
+	sleep 5
 
 }
 
 function start_stream() {
-    # Start streaming the test file over RTMP to the local owncast instance.
+	# Start streaming the test file over RTMP to the local owncast instance.
-    ../../ocTestStream.sh &
+	../../ocTestStream.sh &
-    STREAM_PID=$!
+	STREAM_PID=$!
 
-    echo "Waiting for stream to start..."
+	echo "Waiting for stream to start..."
-    sleep 12
+	sleep 12
 }
 
 function update_storage_config() {
-    echo "Configuring external storage to use ${S3_BUCKET}..."
+	echo "Configuring external storage to use ${S3_BUCKET}..."
 
-    # Hard-coded to admin:abc123 for auth
+	# Hard-coded to admin:abc123 for auth
-    curl --fail 'http://localhost:8080/api/admin/config/s3' \
+	curl --fail 'http://localhost:8080/api/admin/config/s3' \
-    -H 'Authorization: Basic YWRtaW46YWJjMTIz' \
+		-H 'Authorization: Basic YWRtaW46YWJjMTIz' \
-    --data-raw "{\"value\":{\"accessKey\":\"${S3_ACCESS_KEY}\",\"acl\":\"\",\"bucket\":\"${S3_BUCKET}\",\"enabled\":true,\"endpoint\":\"${S3_ENDPOINT}\",\"region\":\"${S3_REGION}\",\"secret\":\"${S3_SECRET}\",\"servingEndpoint\":\"\"}}"
+		--data-raw "{\"value\":{\"accessKey\":\"${S3_ACCESS_KEY}\",\"acl\":\"\",\"bucket\":\"${S3_BUCKET}\",\"enabled\":true,\"endpoint\":\"${S3_ENDPOINT}\",\"region\":\"${S3_REGION}\",\"secret\":\"${S3_SECRET}\",\"servingEndpoint\":\"\"}}"
 }
 
 function kill_with_kids() {
-    # kill a process and all its children (by pid)! return no error.
+	# kill a process and all its children (by pid)! return no error.
-
+
-    if [[ -n $1 ]]; then
+	if [[ -n $1 ]]; then
-        mapfile -t CHILDREN_PID_LIST < <(ps --ppid "$1" -o pid= &>/dev/null || true)
+		mapfile -t CHILDREN_PID_LIST < <(ps --ppid "$1" -o pid= &>/dev/null || true)
-        for child_pid in "${CHILDREN_PID_LIST[@]}"; do
+		for child_pid in "${CHILDREN_PID_LIST[@]}"; do
-            kill "$child_pid" &>/dev/null || true
+			kill "$child_pid" &>/dev/null || true
-            wait "$child_pid" &>/dev/null || true
+			wait "$child_pid" &>/dev/null || true
-        done
+		done
-        kill "$1" &>/dev/null || true
+		kill "$1" &>/dev/null || true
-        wait "$1" &>/dev/null || true
+		wait "$1" &>/dev/null || true
-    fi
+	fi
 }
 
 function finish() {
-    echo "Cleaning up..."
+	echo "Cleaning up..."
-    kill_with_kids "$STREAM_PID"
+	kill_with_kids "$STREAM_PID"
-    kill "$SERVER_PID" &>/dev/null || true
+	kill "$SERVER_PID" &>/dev/null || true
-    wait "$SERVER_PID" &>/dev/null || true
+	wait "$SERVER_PID" &>/dev/null || true
 	rm -fr "$TEMP_DB"
 }
 

utils/utils.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"math/rand"
 	"net/url"
 	"os"
@@ -67,7 +68,7 @@ func Copy(source, destination string) error {
 func Move(source, destination string) error {
 	err := os.Rename(source, destination)
 	if err != nil {
-		log.Warnln("Moving with os.Rename failed, falling back to copy and delete!", err)
+		log.Debugln("Moving with os.Rename failed, falling back to copy and delete!", err)
 		return moveFallback(source, destination)
 	}
 	return nil
@@ -303,10 +304,12 @@ func VerifyFFMpegPath(path string) error {
 }
 
 // CleanupDirectory removes the directory and makes it fresh again. Throws fatal error on failure.
-func CleanupDirectory(path string) {
+func CleanupDirectory(path string, keepOldFiles bool) {
-	log.Traceln("Cleaning", path)
+	if !keepOldFiles {
-	if err := os.RemoveAll(path); err != nil {
+		log.Traceln("Cleaning", path)
-		log.Fatalln("Unable to remove directory. Please check the ownership and permissions", err)
+		if err := os.RemoveAll(path); err != nil {
+			log.Fatalln("Unable to remove directory. Please check the ownership and permissions", err)
+		}
 	}
 	if err := os.MkdirAll(path, 0o750); err != nil {
 		log.Fatalln("Unable to create directory. Please check the ownership and permissions", err)
@@ -448,3 +451,23 @@ func DecodeBase64Image(url string) (bytes []byte, extension string, err error) {
 
 	return bytes, extension, nil
 }
+
+// RoundUpToNearest rounds up to the nearest number divisible.
+func RoundUpToNearest(x float32, to int) int {
+	xInt := int(math.Ceil(float64(x)))
+
+	if xInt%to == 0 {
+		return xInt
+	}
+	return xInt + to - xInt%to
+}
+
+// RoundDownToNearest rounds down to the nearest number divisible.
+func RoundDownToNearest(x float32, to int) int {
+	xInt := int(math.Floor(float64(x)))
+
+	if xInt%to == 0 {
+		return xInt
+	}
+	return xInt - xInt%to
+}