feat(stream): implement section saving

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
+}

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
+}

bruno/VoD-System/Streams/folder.bru

@@ -0,0 +1,14 @@
+meta {
+  name: Streams
+  seq: 7
+}
+
+auth {
+  mode: apikey
+}
+
+auth:apikey {
+  key: 
+  value: 
+  placement: header
+}

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

docker-compose.yml

@@ -22,6 +22,7 @@ services:
       - .env
     volumes:
       - ${MEDIA_DATA}:/app/data/media
+      - ${STREAM_DATA}:/stream
     ports:
       - "8080:8080"
     restart: unless-stopped

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);
+    }
 }

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<APIResponse<Void>> 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<APIResponse<List<Stream>>> getStreamHistory(@PathVariable Long userId) {
         List<Stream> history = streamService.getStreamHistory(userId);

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) {}

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<Stream, Long> {
     Optional<Stream> findByUserAndEndDateIsNull(User user);
     List<Stream> findByUser(User user);
+    List<Stream> findByEndDateIsNullAndLastSeenBefore(LocalDateTime cutoff);
 }

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");

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<Stream> 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));

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<File> saveSection(
+            String streamDirectory,
+            Instant startTime,
+            Instant endTime
+    ) throws IOException, InterruptedException {
+        List<File> 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<String> 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<File> 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;
+    }
+}

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