package com.ddf.vodsystem.services.media; import com.ddf.vodsystem.dto.CommandOutput; import com.ddf.vodsystem.dto.ProgressTracker; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.time.Instant; import java.util.ArrayList; import java.util.Comparator; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import java.util.stream.Stream; /** * Service responsible for performing stream-related media actions, such as * extracting and saving a trimmed section of video from a sequence of * recorded segments. *

* Operations in this service are backed by {@code ffmpeg} and run * asynchronously on a dedicated executor so that potentially long-running * encoding/concatenation work does not block request-handling threads. */ @Service public class StreamActionsService { private static final Logger logger = LoggerFactory.getLogger(StreamActionsService.class); private final CommandRunner commandRunner; private final MetadataService metadataService; public StreamActionsService(CommandRunner commandRunner, MetadataService metadataService) { this.commandRunner = commandRunner; this.metadataService = metadataService; } /** * Concatenates a list of video segments and extracts a trimmed section * from the result, writing the output to {@code outputFile}. *

* Failures (process launch errors, ffmpeg failures, interruption) are * surfaced via the returned future rather than thrown, in line with * normal {@link CompletableFuture} conventions — callers should use * {@code .exceptionally()} / {@code .handle()} rather than a try/catch. * * @param segments the ordered list of segment files to concatenate, * in playback order * @param trimOffset the offset, in seconds, from the start of the * concatenated stream at which to begin the trimmed * section * @param duration the duration, in seconds, of the trimmed section * to extract * @param outputFile the destination file the trimmed section will be * written to; overwritten if it already exists * @return a {@link CompletableFuture} that completes with the ffmpeg * command output on success, or completes exceptionally on failure */ @Async("ffmpegTaskExecutor") public CompletableFuture saveSection( List segments, float trimOffset, float duration, Path outputFile, ProgressTracker progressTracker ) { if (segments.isEmpty()) { return CompletableFuture.failedFuture(new IllegalArgumentException("Segment list cannot be empty")); } if (trimOffset < 0 || duration <= 0) { return CompletableFuture.failedFuture(new IllegalArgumentException( "trimOffset must be >= 0 and duration must be > 0 (got " + trimOffset + ", " + duration + ")")); } Path headSegment = null; try { // Unique per invocation so concurrent saves by different users never // collide on the same head fragment; cleaned up in the finally block. headSegment = Files.createTempFile("reencode-section-", ".ts"); encodeHead(segments.getFirst(), trimOffset, headSegment); String concatInput = Stream.concat(Stream.of(headSegment), segments.stream().skip(1)) .map(p -> p.toAbsolutePath().toString()) .collect(Collectors.joining("|")); List command = List.of( "ffmpeg", "-y", "-i", "concat:" + concatInput, "-t", String.valueOf(duration), "-c", "copy", outputFile.toAbsolutePath().toString() ); logger.info("Saving section with ({} segments) to '{}'", segments.size(), outputFile); CommandOutput output = commandRunner.run(command, line -> commandRunner.setProgress(line, progressTracker, duration)); progressTracker.markComplete(); return CompletableFuture.completedFuture(output); } catch (IOException e) { logger.error("IO error on saveSection call: {}", e.toString()); return CompletableFuture.failedFuture(e); } catch (InterruptedException e) { logger.error("Thread error on saveSection call: {}", e.toString()); Thread.currentThread().interrupt(); return CompletableFuture.failedFuture(e); } catch (RuntimeException e) { logger.error("Failed to save section to '{}': {}", outputFile, e.toString()); return CompletableFuture.failedFuture(e); } finally { if (headSegment != null) { try { Files.deleteIfExists(headSegment); } catch (IOException e) { logger.warn("Failed to delete temp head fragment '{}': {}", headSegment, e.toString()); } } } } public List getSegmentsInRange(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { long startMs = startTime.toEpochMilli(); long endMs = endTime.toEpochMilli(); if (endTime.toEpochMilli() <= startTime.toEpochMilli()) { throw new IllegalArgumentException("Start time must be before end time"); } List results = new ArrayList<>(); try (Stream files = Files.list(streamDirectory)) { List streamFiles = files .filter(p -> p.getFileName().toString().endsWith(".ts")) .filter(Files::isRegularFile) .sorted(Comparator.comparingLong(this::parseTimestampMs)) .toList(); for (int i = 0; i < streamFiles.size(); i++) { Path streamFile = streamFiles.get(i); long segmentStart = parseTimestampMs(streamFile); long segmentEnd = (i + 1 < streamFiles.size()) ? parseTimestampMs(streamFiles.get(i + 1)) : Long.MAX_VALUE; if (segmentStart < endMs && segmentEnd > startMs) { results.add(streamFile); } } } return results; } /** * The segments backing a save and the offset into the first one at which * the requested section begins. * * @param segments the ordered segments to concatenate, in playback order * @param trimOffset the offset, in seconds, into {@code segments.getFirst()} * at which the requested section starts */ public record SegmentSelection(List segments, float trimOffset) {} /** * Selects the recorded {@code .ts} segments overlapping {@code [startTime, endTime)} * and computes the trim offset into the first segment. *

* If {@code startTime} falls in a stream gap after the earliest overlapping segment * (so that segment holds no footage for the requested start), the segment is dropped * and the selection begins at the first genuinely-recorded footage instead. * * @throws IllegalArgumentException if no segments hold footage for the requested range * @throws IOException if listing the stream directory or probing a segment fails */ public SegmentSelection selectSegments(Path streamDirectory, Instant startTime, Instant endTime) throws IOException { List segments = getSegmentsInRange(streamDirectory, startTime, endTime); if (segments.isEmpty()) { throw new IllegalArgumentException("No stream segments found in the given time range"); } long firstMs = parseTimestampMs(segments.getFirst()); float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstMs) / 1000f); // if the first segment is longer than the trim-offset, then ggs if (trimOffset > 0 && probeDurationSeconds(segments.getFirst()) < trimOffset) { segments.removeFirst(); if (segments.isEmpty()) { throw new IllegalArgumentException("No recorded footage at the requested start time"); } trimOffset = 0f; } return new SegmentSelection(segments, trimOffset); } private float probeDurationSeconds(Path segment) throws IOException { try { return metadataService.getVideoMetadata(segment).get().getDuration(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException("Interrupted while probing segment duration: " + segment, e); } catch (ExecutionException e) { throw new IOException("Failed to probe segment duration: " + segment, e); } } public long parseTimestampMs(Path path) { String fullName = path.getFileName().toString(); if (!fullName.endsWith(".ts")) { throw new IllegalArgumentException("File must end in ts"); } String name = fullName.substring(0, fullName.length() - 3); 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; } /** * Re-encodes {@code firstSegment} from {@code trimOffset} to its end, * writing an MPEG-TS fragment that can be stream-copy concatenated with * the untouched fragments that follow it. */ private void encodeHead(Path firstSegment, float trimOffset, Path head) throws IOException, InterruptedException { List command = List.of( "ffmpeg", "-y", "-i", firstSegment.toAbsolutePath().toString(), "-ss", String.valueOf(trimOffset), "-c:v", "libx264", "-preset", "veryfast", "-crf", "20", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "160k", "-muxdelay", "0", "-muxpreload", "0", "-f", "mpegts", head.toAbsolutePath().toString() ); logger.debug("Re-encoding head fragment '{}' from {}s", firstSegment, trimOffset); commandRunner.run(command, line -> { /* head pass is short; not worth reporting */ }); if (!Files.exists(head) || Files.size(head) == 0) { throw new IOException("Head re-encode produced no output — trimOffset " + trimOffset + "s may exceed the duration of " + firstSegment); } } }