dylan/vod-system
1
1
Fork 0
Code Issues 16 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(streaming): timestamp based stream markers #6

Merged
dylan merged 7 commits from feat/stream-markers into master 2026-07-02 19:14:43 +01:00
Conversation 0 Commits 7 Files Changed 25 +407 -5
25 changed files with 407 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example
View File

@@ -16,3 +16,5 @@ POSTGRES_DB=vodSystem
FRONTEND_URL=
BACKEND_URL=
STREAM_RING_LENGTH_SECOND=28800

22
bruno/VoD-System/Markers/Create Marker.bru Normal file
View File

@@ -0,0 +1,22 @@
meta {
name: Create Marker
type: http
seq: 3
}
post {
url: {{base_url}}/markers
body: json
auth: inherit
}
body:json {
{
"message": "Nice play"
}
}
settings {
encodeUrl: true
timeout: 0
}

16
bruno/VoD-System/Markers/Get Marker.bru Normal file
View File

@@ -0,0 +1,16 @@
meta {
name: Get Marker
type: http
seq: 1
}
get {
url: {{base_url}}/markers/5
body: none
auth: inherit
}
settings {
encodeUrl: true
timeout: 0
}

11
bruno/VoD-System/Markers/Get all markers.bru Normal file
View File

@@ -0,0 +1,11 @@
meta {
name: Get all markers
type: http
seq: 2
}
get {
url: {{base_url}}/markers
body: none
auth: inherit
}

8
bruno/VoD-System/Markers/folder.bru Normal file
View File

@@ -0,0 +1,8 @@
meta {
name: Markers
seq: 6
}
auth {
mode: inherit
}

24
bruno/VoD-System/Media/Save Stream Section By Markers.bru Normal file
View File

@@ -0,0 +1,24 @@
meta {
name: Save Stream Section By Markers
type: http
seq: 4
}
post {
url: {{base_url}}/media/save/markers
body: json
auth: inherit
}
body:json {
{
"startMarkerId": 1,
"endMarkerId": 2,
"title": "Hello",
"description": "my clip"
}
}
script:post-response {
bru.setEnvVar("uuid", res.getBody().data.uuid);
}

2
docker-compose.dev.yml
View File

@@ -22,6 +22,8 @@ services:
container_name: vod_rtmp
extra_hosts:
- "host.docker.internal:host-gateway"
env_file:
- .env
environment:
- WEB_SERVER_URL=http://host.docker.internal:8080
ports:

2
docker-compose.yml
View File

@@ -45,6 +45,8 @@ services:
context: ./nginx
dockerfile: Dockerfile
container_name: vod_rtmp
env_file:
- .env
environment:
- WEB_SERVER_URL=http://backend:8080
ports:

2
nginx/default.conf.template
View File

@@ -20,7 +20,7 @@ rtmp {
hls_path /stream;
hls_fragment_naming system;
hls_fragment 3;
hls_playlist_length 28800;
hls_playlist_length ${STREAM_RING_LENGTH_SECOND};
on_publish ${WEB_SERVER_URL}/api/v1/stream/start;
on_publish_done ${WEB_SERVER_URL}/api/v1/stream/stop;

2
nginx/docker-entrypoint.sh
View File

@@ -1,7 +1,7 @@
#!/bin/sh
# must only run on certain env files, otherwise it replaces too much
envsubst '${WEB_SERVER_URL}' \
envsubst '${WEB_SERVER_URL} ${STREAM_RING_LENGTH_SECOND}' \
< /etc/nginx/templates/default.conf.template \
> /etc/nginx/nginx.conf

14
src/main/java/com/ddf/vodsystem/controllers/GlobalExceptionHandler.java
View File

@@ -108,4 +108,18 @@ public class GlobalExceptionHandler {
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.NOT_FOUND).body(response);
}
@ExceptionHandler(NotStreaming.class)
public ResponseEntity<APIResponse<Void>> handleNotStreaming(NotStreaming ex) {
logger.error("NotStreaming: {}", ex.getMessage());
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.CONFLICT).body(response);
}
@ExceptionHandler(MarkerNotFound.class)
public ResponseEntity<APIResponse<Void>> handleMarkerNotFound(MarkerNotFound ex) {
logger.error("MarkerNotFound: {}", ex.getMessage());
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.NOT_FOUND).body(response);
}
}

63
src/main/java/com/ddf/vodsystem/controllers/MarkerController.java Normal file
View File

@@ -0,0 +1,63 @@
package com.ddf.vodsystem.controllers;
import com.ddf.vodsystem.controllers.dto.MarkerCreateRequest;
import com.ddf.vodsystem.controllers.dto.MarkerResponse;
import com.ddf.vodsystem.dto.APIResponse;
import com.ddf.vodsystem.entities.Marker;
import com.ddf.vodsystem.services.MarkerService;
import jakarta.validation.Valid;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.List;
@RestController
@RequestMapping("/api/v1/markers")
public class MarkerController {
private static final String SUCCESS = "success";
private final MarkerService markerService;
public MarkerController(MarkerService markerService) {
this.markerService = markerService;
}
@GetMapping("")
public ResponseEntity<APIResponse<List<MarkerResponse>>> getMarkers() {
List<MarkerResponse> markers = markerService.getUserMarkers().stream()
.map(this::convertToDTO)
.toList();
return ResponseEntity.ok(
new APIResponse<>(SUCCESS, "Markers retrieved successfully", markers)
);
}
@GetMapping("/{id}")
public ResponseEntity<APIResponse<MarkerResponse>> getMarkerById(@PathVariable Long id) {
Marker marker = markerService.getMarkerById(id);
return ResponseEntity.ok(
new APIResponse<>(SUCCESS, "Marker retrieved successfully", convertToDTO(marker))
);
}
@PostMapping("")
public ResponseEntity<APIResponse<MarkerResponse>> createMarker(@Valid @RequestBody MarkerCreateRequest request) {
Marker marker = markerService.create(request.message());
return ResponseEntity.ok(
new APIResponse<>(SUCCESS, "Marker created successfully", convertToDTO(marker))
);
}
private MarkerResponse convertToDTO(Marker marker) {
return new MarkerResponse(
marker.getId(),
marker.getUser().getId(),
marker.getStream().getId(),
marker.getMessage(),
marker.getTimestamp()
);
}
}

16
src/main/java/com/ddf/vodsystem/controllers/MediaController.java
View File

@@ -1,6 +1,7 @@
package com.ddf.vodsystem.controllers;
import com.ddf.vodsystem.controllers.dto.ClipSectionRequest;
import com.ddf.vodsystem.controllers.dto.SaveSectionByMarkerRequest;
import com.ddf.vodsystem.controllers.dto.SaveSectionRequest;
import com.ddf.vodsystem.controllers.dto.UUIDResponse;
import com.ddf.vodsystem.dto.Job;
@@ -54,6 +55,21 @@ public class MediaController {
));
}
@PostMapping("/save/markers")
public ResponseEntity<APIResponse<UUIDResponse>> saveByMarkers(
@Valid @RequestBody SaveSectionByMarkerRequest saveSectionByMarkerRequest) throws IOException {
Job job = mediaService.saveSection(
saveSectionByMarkerRequest.startMarkerId(),
saveSectionByMarkerRequest.endMarkerId(),
saveSectionByMarkerRequest.title(),
saveSectionByMarkerRequest.description());
return ResponseEntity.ok(new APIResponse<>(
SUCCESS,
"Section saving successfully started",
new UUIDResponse(job.getUuid())
));
}
@PostMapping("/clip")
public ResponseEntity<APIResponse<UUIDResponse>> clip(
@RequestBody ClipSectionRequest clipSectionRequest) throws IOException {

7
src/main/java/com/ddf/vodsystem/controllers/dto/MarkerCreateRequest.java Normal file
View File

@@ -0,0 +1,7 @@
package com.ddf.vodsystem.controllers.dto;
import jakarta.validation.constraints.NotBlank;
public record MarkerCreateRequest(
@NotBlank String message
) {}

11
src/main/java/com/ddf/vodsystem/controllers/dto/MarkerResponse.java Normal file
View File

@@ -0,0 +1,11 @@
package com.ddf.vodsystem.controllers.dto;
import java.time.Instant;
public record MarkerResponse(
Long id,
Long userId,
Long streamId,
String message,
Instant timestamp
) {}

9
src/main/java/com/ddf/vodsystem/controllers/dto/SaveSectionByMarkerRequest.java Normal file
View File

@@ -0,0 +1,9 @@
package com.ddf.vodsystem.controllers.dto;
import jakarta.validation.constraints.NotNull;
public record SaveSectionByMarkerRequest(
@NotNull Long startMarkerId,
@NotNull Long endMarkerId,
String title,
String description) {}

33
src/main/java/com/ddf/vodsystem/entities/Marker.java Normal file
View File

@@ -0,0 +1,33 @@
package com.ddf.vodsystem.entities;
import com.fasterxml.jackson.annotation.JsonIgnore;
import jakarta.persistence.*;
import lombok.Data;
import java.time.Instant;
@Entity
@Table(name = "markers")
@Data
public class Marker {
@Id
@Column(name = "id", nullable = false, unique = true)
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@ManyToOne(fetch = FetchType.LAZY)
@JsonIgnore
@JoinColumn(name = "user_id", nullable = false)
private User user;
@ManyToOne(fetch = FetchType.LAZY)
@JsonIgnore
@JoinColumn(name = "stream_id", nullable = false)
private Stream stream;
@Column(name = "message", length = 255)
private String message;
@Column(name = "timestamp")
private Instant timestamp;
}

7
src/main/java/com/ddf/vodsystem/exceptions/MarkerNotFound.java Normal file
View File

@@ -0,0 +1,7 @@
package com.ddf.vodsystem.exceptions;
public class MarkerNotFound extends RuntimeException {
public MarkerNotFound(String message) {
super(message);
}
}

7
src/main/java/com/ddf/vodsystem/exceptions/NotStreaming.java Normal file
View File

@@ -0,0 +1,7 @@
package com.ddf.vodsystem.exceptions;
public class NotStreaming extends RuntimeException {
public NotStreaming(String message) {
super(message);
}
}

19
src/main/java/com/ddf/vodsystem/repositories/MarkerRepository.java Normal file
View File

@@ -0,0 +1,19 @@
package com.ddf.vodsystem.repositories;
import com.ddf.vodsystem.entities.Marker;
import com.ddf.vodsystem.entities.User;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Query;
import org.springframework.stereotype.Repository;
import java.time.Instant;
import java.util.List;
@Repository
public interface MarkerRepository extends JpaRepository<Marker, Long> {
@Query("SELECT m FROM Marker m WHERE m.user = ?1")
List<Marker> findByUser(User user);
@Query("SELECT m FROM Marker m WHERE m.timestamp <= ?1")
List<Marker> findAllBefore(Instant time);
}

2
src/main/java/com/ddf/vodsystem/services/ClipService.java
View File

@@ -174,7 +174,7 @@ public class ClipService {
Path file = directoryService.resolvePath(path);
if (!Files.exists(file)) {
throw new JobNotFound("Thumbnail file not found");
throw new ClipNotFound("Thumbnail file not found");
}
return new FileSystemResource(file);

82
src/main/java/com/ddf/vodsystem/services/MarkerService.java Normal file
View File

@@ -0,0 +1,82 @@
package com.ddf.vodsystem.services;
import com.ddf.vodsystem.entities.Marker;
import com.ddf.vodsystem.entities.Stream;
import com.ddf.vodsystem.entities.User;
import com.ddf.vodsystem.exceptions.MarkerNotFound;
import com.ddf.vodsystem.exceptions.NotAuthenticated;
import com.ddf.vodsystem.exceptions.NotStreaming;
import com.ddf.vodsystem.repositories.MarkerRepository;
import jakarta.transaction.Transactional;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Instant;
import java.util.List;
import java.util.Optional;
@Service
public class MarkerService {
private final UserService userService;
private final StreamService streamService;
private final MarkerRepository markerRepository;
@Value("${stream.max-length}")
private Long maxStreamLength;
public MarkerService(UserService userService, StreamService streamService, MarkerRepository markerRepository) {
this.userService = userService;
this.streamService = streamService;
this.markerRepository = markerRepository;
}
public Marker getMarkerById(Long id) {
Optional<Marker> marker = markerRepository.findById(id);
if (marker.isEmpty()) {
throw new MarkerNotFound("Marker does not exist");
}
Optional<User> user = userService.getLoggedInUser();
if (user.isEmpty() || !user.get().equals(marker.get().getUser())) {
throw new NotAuthenticated("User not authenticated for this marker");
}
return marker.get();
}
public List<Marker> getUserMarkers() {
Optional<User> user = userService.getLoggedInUser();
if (user.isEmpty()) {
throw new NotAuthenticated("Must be logged in to get markers");
}
return markerRepository.findByUser(user.get());
}
public Marker create(String message) {
Optional<Stream> stream = streamService.getActiveStream();
if (stream.isEmpty()) {
throw new NotStreaming("User must be streaming to mark");
}
Marker marker = new Marker();
marker.setStream(stream.get());
marker.setUser(stream.get().getUser());
marker.setMessage(message);
marker.setTimestamp(Instant.now());
return markerRepository.saveAndFlush(marker);
}
@Scheduled(fixedDelay = 360_000)
@Transactional
public void deleteOldMarkers() {
Instant cutoff = Instant.now().minusSeconds(maxStreamLength);
List<Marker> stale = markerRepository.findAllBefore(cutoff);
markerRepository.deleteAllInBatch(stale);
}
}

37
src/main/java/com/ddf/vodsystem/services/MediaService.java
View File

@@ -3,7 +3,9 @@ package com.ddf.vodsystem.services;
import com.ddf.vodsystem.dto.ClipOptions;
import com.ddf.vodsystem.dto.Job;
import com.ddf.vodsystem.dto.JobState;
import com.ddf.vodsystem.entities.Marker;
import com.ddf.vodsystem.entities.User;
import com.ddf.vodsystem.exceptions.MarkerNotFound;
import com.ddf.vodsystem.exceptions.NotAuthenticated;
import com.ddf.vodsystem.services.media.CompressionService;
import com.ddf.vodsystem.services.media.StreamActionsService;
@@ -32,12 +34,13 @@ public class MediaService {
private static final int HLS_SEGMENT_LENGTH = 3;
private final StreamActionsService streamActionsService;
private final VodService vodService;
private final MarkerService markerService;
public MediaService(JobRegistryService jobRegistryService,
DirectoryService directoryService,
CompressionService compressionService,
UserService userService,
ClipService clipService, StreamActionsService streamActionsService, VodService vodService) {
ClipService clipService, StreamActionsService streamActionsService, VodService vodService, MarkerService markerService) {
this.jobRegistryService = jobRegistryService;
this.directoryService = directoryService;
this.compressionService = compressionService;
@@ -45,6 +48,7 @@ public class MediaService {
this.clipService = clipService;
this.streamActionsService = streamActionsService;
this.vodService = vodService;
this.markerService = markerService;
}
/**
@@ -149,6 +153,37 @@ public class MediaService {
);
}
/**
* Saves a section of the currently authenticated user's live stream between two markers.
* <p>
* Resolves {@code startMarkerId} and {@code endMarkerId} to their marker timestamps and
* delegates to {@link #saveSection(Instant, Instant, String, String)}.
*
* @param startMarkerId id of the marker whose timestamp begins the section to save
* @param endMarkerId id of the marker whose timestamp ends the section to save
* @param title title for the saved VoD; defaults to the current timestamp if {@code null}
* @param description description for the saved VoD; defaults to an empty string if {@code null}
* @return the {@link Job} tracking the save; state transitions to
* {@link JobState#SUCCEEDED} or {@link JobState#FAILED} asynchronously
* @throws MarkerNotFound if either marker id does not exist
* @throws NotAuthenticated if no user is currently authenticated, or the caller does not
* own one of the referenced markers
* @throws IllegalArgumentException if the start marker's timestamp is not before the end
* marker's, or if no stream segments exist in the given range
* @throws IOException if reading the stream directory or its segments fails
*/
public Job saveSection(
Long startMarkerId,
Long endMarkerId,
String title,
String description
) throws IOException {
Marker start = markerService.getMarkerById(startMarkerId);
Marker end = markerService.getMarkerById(endMarkerId);
return saveSection(start.getTimestamp(), end.getTimestamp(), title, description);
}
/**
* Saves the last {@code duration} seconds of the currently authenticated user's live stream as a clip.
* <p>

1
src/main/resources/application.properties
View File

@@ -3,6 +3,7 @@ spring.application.name=vodSystem
# VODs
spring.servlet.multipart.max-file-size=2GB
spring.servlet.multipart.max-request-size=2GB
stream.max-length=${STREAM_RING_LENGTH_SECOND}
# Database
spring.datasource.url=${POSTGRES_URL}

11
src/main/resources/db/schema.sql
View File

@@ -55,3 +55,14 @@ CREATE TABLE IF NOT EXISTS vods (
thumbnail_path VARCHAR(255) NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS markers (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL,
stream_id BIGINT NOT NULL,
message VARCHAR(255),
timestamp TIMESTAMPTZ DEFAULT NOW(),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (stream_id) REFERENCES streams(id) ON DELETE CASCADE
);