From bf59cb087f6debcb14a82a8d4358f924383db80b Mon Sep 17 00:00:00 2001 From: Dylan De Faoite Date: Sat, 6 Jun 2026 15:22:36 +0100 Subject: [PATCH] feat(stream): implement section saving --- bruno/VoD-System/Streams/History Streams.bru | 20 +++++ bruno/VoD-System/Streams/Save Section.bru | 27 ++++++ bruno/VoD-System/Streams/folder.bru | 14 +++ docker-compose.dev.yml | 1 + docker-compose.yml | 1 + .../vodsystem/configuration/AsyncConfig.java | 13 ++- .../controllers/StreamController.java | 9 ++ .../ddf/vodsystem/dto/SaveSectionRequest.java | 5 ++ .../repositories/StreamRepository.java | 2 + .../vodsystem/services/DirectoryService.java | 13 +++ .../ddf/vodsystem/services/StreamService.java | 51 ++++++++++- .../services/media/StreamActionsService.java | 90 +++++++++++++++++++ src/main/resources/application.properties | 3 + 13 files changed, 246 insertions(+), 3 deletions(-) create mode 100644 bruno/VoD-System/Streams/History Streams.bru create mode 100644 bruno/VoD-System/Streams/Save Section.bru create mode 100644 bruno/VoD-System/Streams/folder.bru create mode 100644 src/main/java/com/ddf/vodsystem/dto/SaveSectionRequest.java create mode 100644 src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java diff --git a/bruno/VoD-System/Streams/History Streams.bru b/bruno/VoD-System/Streams/History Streams.bru new file mode 100644 index 0000000..3a00096 --- /dev/null +++ b/bruno/VoD-System/Streams/History Streams.bru @@ -0,0 +1,20 @@ +meta { + name: History Streams + type: http + seq: 1 +} + +get { + url: {{base_url}}/stream/history/1 + body: none + auth: bearer +} + +auth:bearer { + token: {{token}} +} + +settings { + encodeUrl: true + timeout: 0 +} diff --git a/bruno/VoD-System/Streams/Save Section.bru b/bruno/VoD-System/Streams/Save Section.bru new file mode 100644 index 0000000..fdbee68 --- /dev/null +++ b/bruno/VoD-System/Streams/Save Section.bru @@ -0,0 +1,27 @@ +meta { + name: Save Section + type: http + seq: 2 +} + +post { + url: {{base_url}}/stream/save + body: json + auth: bearer +} + +auth:bearer { + token: {{token}} +} + +body:json { + { + "startTime": "2026-06-06T13:56:00Z", + "endTime": "2026-06-06T13:58:00Z" + } +} + +settings { + encodeUrl: true + timeout: 0 +} diff --git a/bruno/VoD-System/Streams/folder.bru b/bruno/VoD-System/Streams/folder.bru new file mode 100644 index 0000000..6b8f9c2 --- /dev/null +++ b/bruno/VoD-System/Streams/folder.bru @@ -0,0 +1,14 @@ +meta { + name: Streams + seq: 7 +} + +auth { + mode: apikey +} + +auth:apikey { + key: + value: + placement: header +} \ No newline at end of file diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 6cc3fbf..da17515 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -26,6 +26,7 @@ services: - WEB_SERVER_URL=http://host.docker.internal:8080 ports: - "1935:1935" + - "8088:8088" volumes: - ${STREAM_DATA}:/stream pull_policy: build diff --git a/docker-compose.yml b/docker-compose.yml index 4fa72be..3ce1e7f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -22,6 +22,7 @@ services: - .env volumes: - ${MEDIA_DATA}:/app/data/media + - ${STREAM_DATA}:/stream ports: - "8080:8080" restart: unless-stopped diff --git a/src/main/java/com/ddf/vodsystem/configuration/AsyncConfig.java b/src/main/java/com/ddf/vodsystem/configuration/AsyncConfig.java index ba7a39e..b6200ac 100644 --- a/src/main/java/com/ddf/vodsystem/configuration/AsyncConfig.java +++ b/src/main/java/com/ddf/vodsystem/configuration/AsyncConfig.java @@ -1,13 +1,18 @@ package com.ddf.vodsystem.configuration; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.AsyncConfigurer; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor; @Configuration -public class AsyncConfig { +public class AsyncConfig implements AsyncConfigurer { + private static final Logger logger = LoggerFactory.getLogger(AsyncConfig.class); @Bean(name = "ffmpegTaskExecutor") public Executor taskExecutor() { @@ -19,4 +24,10 @@ public class AsyncConfig { executor.initialize(); return executor; } + + @Override + public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { + return (ex, method, params) -> + logger.error("Async task '{}' failed: {}", method.getName(), ex.getMessage(), ex); + } } diff --git a/src/main/java/com/ddf/vodsystem/controllers/StreamController.java b/src/main/java/com/ddf/vodsystem/controllers/StreamController.java index 819ee60..74cb465 100644 --- a/src/main/java/com/ddf/vodsystem/controllers/StreamController.java +++ b/src/main/java/com/ddf/vodsystem/controllers/StreamController.java @@ -1,6 +1,7 @@ package com.ddf.vodsystem.controllers; import com.ddf.vodsystem.dto.APIResponse; +import com.ddf.vodsystem.dto.SaveSectionRequest; import com.ddf.vodsystem.entities.Stream; import com.ddf.vodsystem.services.StreamService; import org.slf4j.Logger; @@ -8,6 +9,7 @@ import org.slf4j.LoggerFactory; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*; +import java.io.IOException; import java.util.List; @RestController @@ -44,6 +46,13 @@ public class StreamController { return ResponseEntity.ok(new APIResponse<>("success", "Heartbeat recorded", null)); } + @PostMapping("/save") + public ResponseEntity> saveSection( + @RequestBody SaveSectionRequest request) throws IOException, InterruptedException { + streamService.saveSection(request.startTime(), request.endTime()); + return ResponseEntity.ok(new APIResponse<>("success", "Section saved", null)); + } + @GetMapping("/history/{userId}") public ResponseEntity>> getStreamHistory(@PathVariable Long userId) { List history = streamService.getStreamHistory(userId); diff --git a/src/main/java/com/ddf/vodsystem/dto/SaveSectionRequest.java b/src/main/java/com/ddf/vodsystem/dto/SaveSectionRequest.java new file mode 100644 index 0000000..0b1eb5e --- /dev/null +++ b/src/main/java/com/ddf/vodsystem/dto/SaveSectionRequest.java @@ -0,0 +1,5 @@ +package com.ddf.vodsystem.dto; + +import java.time.Instant; + +public record SaveSectionRequest(Instant startTime, Instant endTime) {} diff --git a/src/main/java/com/ddf/vodsystem/repositories/StreamRepository.java b/src/main/java/com/ddf/vodsystem/repositories/StreamRepository.java index a494f5f..e1ce754 100644 --- a/src/main/java/com/ddf/vodsystem/repositories/StreamRepository.java +++ b/src/main/java/com/ddf/vodsystem/repositories/StreamRepository.java @@ -5,6 +5,7 @@ import com.ddf.vodsystem.entities.User; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Repository; +import java.time.LocalDateTime; import java.util.List; import java.util.Optional; @@ -12,4 +13,5 @@ import java.util.Optional; public interface StreamRepository extends JpaRepository { Optional findByUserAndEndDateIsNull(User user); List findByUser(User user); + List findByEndDateIsNullAndLastSeenBefore(LocalDateTime cutoff); } diff --git a/src/main/java/com/ddf/vodsystem/services/DirectoryService.java b/src/main/java/com/ddf/vodsystem/services/DirectoryService.java index ce632e5..924a5f7 100644 --- a/src/main/java/com/ddf/vodsystem/services/DirectoryService.java +++ b/src/main/java/com/ddf/vodsystem/services/DirectoryService.java @@ -59,6 +59,19 @@ public class DirectoryService { return file; } + public File getVodFile(String filename) { + String dir = outputDir + File.separator + "vods" + File.separator + filename; + File file = new File(dir); + + try { + createDirectory(file.getParent()); + } catch (IOException e) { + logger.error("Error creating vods directory: {}", e.getMessage()); + } + + return file; + } + public File getUserThumbnailsFile(Long userId, String fileName) { if (userId == null || fileName == null || fileName.isEmpty()) { throw new IllegalArgumentException("User ID and file name cannot be null or empty"); diff --git a/src/main/java/com/ddf/vodsystem/services/StreamService.java b/src/main/java/com/ddf/vodsystem/services/StreamService.java index 23e2d9e..6e67c6b 100644 --- a/src/main/java/com/ddf/vodsystem/services/StreamService.java +++ b/src/main/java/com/ddf/vodsystem/services/StreamService.java @@ -4,21 +4,42 @@ import com.ddf.vodsystem.entities.Stream; import com.ddf.vodsystem.entities.User; import com.ddf.vodsystem.exceptions.AlreadyStreaming; import com.ddf.vodsystem.exceptions.KeyNotFound; +import com.ddf.vodsystem.exceptions.NotAuthenticated; import com.ddf.vodsystem.repositories.StreamRepository; +import com.ddf.vodsystem.services.media.StreamActionsService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import java.io.File; +import java.io.IOException; +import java.time.Instant; import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.List; import java.util.Optional; @Service public class StreamService { + private static final Logger logger = LoggerFactory.getLogger(StreamService.class); + private static final int HEARTBEAT_TIMEOUT_SECONDS = 60; + private final StreamRepository streamRepository; private final UserService userService; + private final StreamActionsService streamActionsService; - public StreamService(StreamRepository streamRepository, UserService userService) { + @Value("${storage.stream}") + private String streamDataPath; + + public StreamService(StreamRepository streamRepository, + UserService userService, + StreamActionsService streamActionsService) { this.streamRepository = streamRepository; this.userService = userService; + this.streamActionsService = streamActionsService; } public Stream startStream(String streamKey) { @@ -43,7 +64,6 @@ public class StreamService { .orElseThrow(() -> new IllegalStateException( "No active stream found for user " + user.getUsername())); - stream.setEndDate(LocalDateTime.now()); streamRepository.saveAndFlush(stream); } @@ -70,6 +90,33 @@ public class StreamService { return streamRepository.findByUser(user); } + public void saveSection(Instant startTime, Instant endTime) throws IOException, InterruptedException { + User user = userService.getLoggedInUser() + .orElseThrow(() -> new NotAuthenticated("User is not authenticated")); + + String streamDirectory = streamDataPath + File.separator + user.getStreamKey(); + streamActionsService.saveSection(streamDirectory, startTime, endTime) + .exceptionally(ex -> { + logger.error("saveSection failed for user {}: {}", user.getUsername(), ex.getMessage(), ex); + return null; + }); + } + + @Scheduled(fixedDelay = 30_000) + @Transactional + public void endStaleStreams() { + LocalDateTime cutoff = LocalDateTime.now().minusSeconds(HEARTBEAT_TIMEOUT_SECONDS); + List stale = streamRepository.findByEndDateIsNullAndLastSeenBefore(cutoff); + if (stale.isEmpty()) return; + + LocalDateTime now = LocalDateTime.now(); + for (Stream stream : stale) { + stream.setEndDate(now); + logger.warn("Stream {} for user {} ended due to heartbeat timeout", stream.getId(), stream.getUser().getUsername()); + } + streamRepository.saveAllAndFlush(stale); + } + private User resolveUser(String streamKey) { return userService.getUserByStreamKey(streamKey) .orElseThrow(() -> new KeyNotFound("Stream key not found: " + streamKey)); diff --git a/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java new file mode 100644 index 0000000..831d992 --- /dev/null +++ b/src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java @@ -0,0 +1,90 @@ +package com.ddf.vodsystem.services.media; + +import com.ddf.vodsystem.services.DirectoryService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Service; + +import java.io.File; +import java.io.IOException; +import java.time.Instant; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +@Service +public class StreamActionsService { + private static final Logger logger = LoggerFactory.getLogger(StreamActionsService.class); + + private final DirectoryService directoryService; + + public StreamActionsService(DirectoryService directoryService) { + this.directoryService = directoryService; + } + + @Async("ffmpegTaskExecutor") + public CompletableFuture saveSection( + String streamDirectory, + Instant startTime, + Instant endTime + ) throws IOException, InterruptedException { + List segments = getSegmentsInRange(streamDirectory, startTime, endTime); + if (segments.isEmpty()) { + throw new IllegalArgumentException("No segments found in range [" + startTime + ", " + endTime + "]"); + } + + long firstSegmentMs = parseTimestampMs(segments.get(0)); + float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstSegmentMs) / 1000f); + float duration = (endTime.toEpochMilli() - startTime.toEpochMilli()) / 1000f; + + String concatInput = segments.stream() + .map(File::getAbsolutePath) + .collect(Collectors.joining("|")); + + File outputFile = directoryService.getVodFile(UUID.randomUUID() + ".mp4"); + + List command = List.of( + "ffmpeg", + "-y", + "-i", "concat:" + concatInput, + "-ss", String.valueOf(trimOffset), + "-t", String.valueOf(duration), + "-c", "copy", + outputFile.getAbsolutePath() + ); + + logger.info("Saving section [{} - {}] from '{}' ({} segments) to '{}'", + startTime, endTime, streamDirectory, segments.size(), outputFile.getAbsolutePath()); + CommandRunner.run(command, line -> logger.debug("{}", line)); + + return CompletableFuture.completedFuture(outputFile); + } + + private List getSegmentsInRange(String streamDirectory, Instant startTime, Instant endTime) { + File dir = new File(streamDirectory); + File[] tsFiles = dir.listFiles((d, name) -> name.endsWith(".ts")); + if (tsFiles == null) return List.of(); + + long startMs = startTime.toEpochMilli(); + long endMs = endTime.toEpochMilli(); + + return Arrays.stream(tsFiles) + .filter(f -> { + long segMs = parseTimestampMs(f); + return segMs < endMs && segMs + 3_000 > startMs; + }) + .sorted(Comparator.comparingLong(this::parseTimestampMs)) + .collect(Collectors.toList()); + } + + private long parseTimestampMs(File file) { + String name = file.getName().replace(".ts", ""); + long value = Long.parseLong(name); + // nginx hls_fragment_naming system uses seconds; convert to ms + return value < 1_000_000_000_000L ? value * 1000L : value; + } +} diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 2f9acc6..1d4fc35 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -28,5 +28,8 @@ jwt.expiration=3600000 # Frontend frontend.url=${FRONTEND_URL} +# Stream HLS +storage.stream=${STREAM_DATA:/stream} + ## Server Configuration server.servlet.session.timeout=30m \ No newline at end of file