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

feat(streaming): add backend streaming support with nginx-rtmp #1

Merged
dylan merged 31 commits from feat/rtmp-streaming into master 2026-06-20 01:33:56 +01:00
Conversation 1 Commits 31 Files Changed 42 +851 -61
42 changed files with 851 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
.env.example
View File

@@ -1,3 +1,18 @@
# Secrets
GOOGLE_CLIENT_ID=
GOOGLE_CLIENT_SECRET=
JWT_SECRET_KEY=
# Volumes
POSTGRES_DATA=
STREAM_DATA=
MEDIA_DATA=
# DB Credentials
POSTGRES_URL=jdbc:postgresql://postgres:5432/vodSystem
POSTGRES_USER=myuser
POSTGRES_PASSWORD=mypassword
POSTGRES_DB=vodSystem
FRONTEND_URL=
BACKEND_URL=

2
.gitignore vendored
View File

@@ -40,5 +40,7 @@ build/
### Generated Files ###
videos/
outputs/
generated
.mvn

19
.mvn/wrapper/maven-wrapper.properties vendored
View File

@@ -1,19 +0,0 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
wrapperVersion=3.3.2
distributionType=only-script
distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.9/apache-maven-3.9.9-bin.zip

20
bruno/VoD-System/Streams/History Streams.bru Normal file
View File

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

22
bruno/VoD-System/Streams/Save Clip.bru Normal file
View File

@@ -0,0 +1,22 @@
meta {
name: Save Clip
type: http
seq: 3
}
post {
url: {{base_url}}/stream/clip
body: json
auth: inherit
}
body:json {
{
"duration": 60
}
}
settings {
encodeUrl: true
timeout: 0
}

23
bruno/VoD-System/Streams/Save Section.bru Normal file
View File

@@ -0,0 +1,23 @@
meta {
name: Save Section
type: http
seq: 2
}
post {
url: {{base_url}}/stream/save
body: json
auth: inherit
}
body:json {
{
"startTime": "2026-06-20T00:15:00Z",
"endTime": "2026-06-20T00:16:00Z"
}
}
settings {
encodeUrl: true
timeout: 0
}

12
bruno/VoD-System/Streams/folder.bru Normal file
View File

@@ -0,0 +1,12 @@
meta {
name: Streams
seq: 7
}
auth {
mode: bearer
}
auth:bearer {
token: {{token}}
}

32
docker-compose.dev.yml Normal file
View File

@@ -0,0 +1,32 @@
# Local dev: only infrastructure services
# Backend and frontend run directly on the host
# Usage: docker compose -f docker-compose.dev.yml up
services:
postgres:
image: postgres:15
container_name: vod_postgres
environment:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: ${POSTGRES_DB}
volumes:
- ${POSTGRES_DATA}:/var/lib/postgresql/data
ports:
- "5432:5432"
rtmp:
build:
context: ./nginx
dockerfile: Dockerfile
container_name: vod_rtmp
extra_hosts:
- "host.docker.internal:host-gateway"
environment:
- WEB_SERVER_URL=http://host.docker.internal:8080
ports:
- "1935:1935"
volumes:
- ${STREAM_DATA}:/stream
pull_policy: build
restart: unless-stopped

36
docker-compose.yml
View File

@@ -1,16 +1,15 @@
services:
postgres:
image: postgres:15
container_name: vod_postgres
environment:
POSTGRES_USER: myuser
POSTGRES_PASSWORD: mypassword
POSTGRES_DB: vodSystem
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: ${POSTGRES_DB}
volumes:
- ${POSTGRES_DATA}:/var/lib/postgresql/data
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
backend:
build:
@@ -21,6 +20,9 @@ services:
- postgres
env_file:
- .env
volumes:
- ${MEDIA_DATA}:/app/data/media
- ${STREAM_DATA}:/stream
ports:
- "8080:8080"
restart: unless-stopped
@@ -30,13 +32,27 @@ services:
context: ./frontend
dockerfile: Dockerfile
container_name: vod_frontend
env_file:
- .env
environment:
VITE_API_URL: ${BACKEND_URL}
VITE_GOOGLE_CLIENT_ID: ${GOOGLE_CLIENT_ID}
depends_on:
- backend
ports:
- "5173:5173"
restart: unless-stopped
volumes:
postgres_data:
rtmp:
build:
context: ./nginx
dockerfile: Dockerfile
container_name: vod_rtmp
environment:
- WEB_SERVER_URL=http://backend:8080
ports:
- "1935:1935"
volumes:
- ${STREAM_DATA}:/stream
pull_policy: build
restart: unless-stopped
depends_on:
- backend

1
frontend/.dockerignore Normal file
View File

@@ -0,0 +1 @@
node_modules

2
frontend/Dockerfile
View File

@@ -2,7 +2,7 @@ FROM node:20-alpine
WORKDIR /app
COPY package*.json ./
RUN npm install
RUN npm ci
COPY . .
EXPOSE 5173

2
frontend/src/App.tsx
View File

@@ -7,6 +7,7 @@ import Home from './pages/Home';
import {useEffect} from "react";
import MyClips from './pages/MyClips';
import VideoPlayer from "./pages/VideoPlayer.tsx";
import Profile from "./pages/Profile.tsx";
import { GoogleOAuthProvider } from '@react-oauth/google';
@@ -25,6 +26,7 @@ function App() {
<Route path="/create/:id" element={<ClipEdit />} />
<Route path="/my-clips" element={<MyClips />} />
<Route path="/video/:id" element={<VideoPlayer />} />
<Route path="/profile" element={<Profile />} />
</Route>
</Routes>
</Router>

19
frontend/src/components/Sidebar.tsx
View File

@@ -1,6 +1,6 @@
import clsx from "clsx";
import SidebarButton from "./buttons/SidebarButton.tsx";
import { Plus, Film, Home } from 'lucide-react';
import { Plus, Film, Home, User as UserIcon } from 'lucide-react';
import Box from "./Box.tsx";
import type {User} from "../utils/types.ts";
@@ -30,11 +30,18 @@ const Sidebar = ({user, className}: props) => {
/>
{ user && (
<SidebarButton
url={"/my-clips"}
logo={<Film size={20}/>}
label={"My Clips"}
/>
<>
<SidebarButton
url={"/my-clips"}
logo={<Film size={20}/>}
label={"My Clips"}
/>
<SidebarButton
url={"/profile"}
logo={<UserIcon size={20}/>}
label={"Profile"}
/>
</>
)}
</Box>
);

2
frontend/src/components/Topbar.tsx
View File

@@ -70,6 +70,8 @@ const Topbar = ({
/>
<Dropdown label={user.name}>
<DropdownItem item="Profile"
onClick={() => navigate("/profile")} />
<DropdownItem item="Logout"
onClick={() => handleLogout()}
className={"text-red-500 font-medium"} />

3
frontend/src/components/video/VideoCard.tsx
View File

@@ -13,6 +13,7 @@ type VideoCardProps = {
}
const fallbackThumbnail = "../../../public/default_thumbnail.png";
const API_URL = import.meta.env.VITE_API_URL;
const VideoCard = ({
id,
@@ -45,7 +46,7 @@ const VideoCard = ({
<div className={clsx("flex flex-col", className)}>
<div className={"relative inline-block"}>
<img
src={thumbnailAvailable ? `/api/v1/download/thumbnail/${id}` : fallbackThumbnail}
src={thumbnailAvailable ? API_URL + `/api/v1/download/thumbnail/${id}` : fallbackThumbnail}
alt="Video Thumbnail"
/>

6
frontend/src/pages/ClipEdit.tsx
View File

@@ -10,10 +10,12 @@ import type { VideoMetadata } from "../utils/types.ts";
import Box from "../components/Box.tsx";
import MetadataBox from "../components/video/MetadataBox.tsx";
const API_URL = import.meta.env.VITE_API_URL;
const ClipEdit = () => {
const { id } = useParams();
const videoRef = useRef<HTMLVideoElement | null>(null);
const videoUrl = `/api/v1/download/input/${id}`
const videoUrl = API_URL + `/api/v1/download/input/${id}`
const [metadata, setMetadata] = useState<VideoMetadata | null>(null);
const [playbackValue, setPlaybackValue] = useState(0);
const [outputMetadata, setOutputMetadata] = useState<VideoMetadata>({
@@ -65,7 +67,7 @@ const ClipEdit = () => {
}
const handleDownload = async () => {
const response = await fetch(`/api/v1/download/output/${id}`);
const response = await fetch(API_URL + `/api/v1/download/output/${id}`);
if (!response.ok) {
console.error('Download failed');

85
frontend/src/pages/Profile.tsx Normal file
View File

@@ -0,0 +1,85 @@
import { useState, useEffect } from "react";
import { Eye, EyeOff, Copy, Check } from "lucide-react";
import Box from "../components/Box.tsx";
import { getUser } from "../utils/endpoints.ts";
import type { User } from "../utils/types.ts";
const Profile = () => {
const [user, setUser] = useState<User | null>(null);
const [keyVisible, setKeyVisible] = useState(false);
const [copied, setCopied] = useState(false);
useEffect(() => {
getUser().then(setUser).catch(console.error);
}, []);
const handleCopy = () => {
if (!user?.streamKey) return;
navigator.clipboard.writeText(user.streamKey);
setCopied(true);
setTimeout(() => setCopied(false), 2000);
};
if (!user) {
return (
<div className="flex justify-center items-center h-full text-gray-500 text-lg">
Please log in to view your profile.
</div>
);
}
return (
<div className="px-8 py-10 max-w-2xl mx-auto">
<h1 className="text-2xl font-semibold text-gray-700 mb-6">Profile</h1>
<Box className="p-6 mb-6 flex items-center gap-6">
<img
src={user.profilePictureUrl}
referrerPolicy="no-referrer"
alt="Profile picture"
className="w-20 h-20 rounded-full shadow"
/>
<div>
<p className="text-xl font-semibold text-gray-800">{user.name}</p>
<p className="text-sm text-gray-500">@{user.username}</p>
<p className="text-sm text-gray-500">{user.email}</p>
<p className="text-xs text-gray-400 mt-1">
Member since {new Date(user.createdAt).toLocaleDateString(undefined, { year: "numeric", month: "long", day: "numeric" })}
</p>
</div>
</Box>
<Box className="p-6">
<h2 className="text-lg font-semibold text-gray-700 mb-1">Stream Key</h2>
<p className="text-sm text-gray-500 mb-4">
Use this key in OBS or any RTMP-compatible software to start streaming.
Keep it private — anyone with this key can stream to your account.
</p>
<div className="flex items-center gap-2">
<div className="flex-1 bg-gray-100 border border-gray-300 rounded-md px-4 py-2 font-mono text-sm text-gray-700 overflow-hidden">
{keyVisible ? user.streamKey : "•".repeat(user.streamKey.length)}
</div>
<button
onClick={() => setKeyVisible(v => !v)}
title={keyVisible ? "Hide key" : "Show key"}
className="p-2 rounded-md bg-gray-300 hover:bg-gray-400 transition-colors text-gray-700"
>
{keyVisible ? <EyeOff size={18} /> : <Eye size={18} />}
</button>
<button
onClick={handleCopy}
title="Copy key"
className="p-2 rounded-md bg-gray-300 hover:bg-gray-400 transition-colors text-gray-700"
>
{copied ? <Check size={18} className="text-green-600" /> : <Copy size={18} />}
</button>
</div>
</Box>
</div>
);
};
export default Profile;

5
frontend/src/utils/types.ts
View File

@@ -17,8 +17,11 @@ type APIResponse = {
type User = {
name: string,
username: string,
email: string,
profilePictureUrl: string
profilePictureUrl: string,
streamKey: string,
createdAt: string,
}
type Clip = {

9
nginx/Dockerfile Normal file
View File

@@ -0,0 +1,9 @@
FROM tiangolo/nginx-rtmp
RUN apt-get update && apt-get install -y gettext-base && rm -rf /var/lib/apt/lists/*
COPY default.conf.template /etc/nginx/templates/default.conf.template
COPY docker-entrypoint.sh /docker-entrypoint.sh
RUN chmod +x /docker-entrypoint.sh
ENTRYPOINT ["/docker-entrypoint.sh"]

48
nginx/default.conf.template Normal file
View File

@@ -0,0 +1,48 @@
worker_processes auto;
events {
worker_connections 1024;
}
rtmp {
server {
listen 1935;
chunk_size 4096;
application live {
live on;
record off;
allow publish all;
allow play all;
hls on;
hls_nested on;
hls_path /stream;
hls_fragment_naming system;
hls_fragment 3;
hls_playlist_length 28800;
on_publish ${WEB_SERVER_URL}/api/v1/stream/start;
on_publish_done ${WEB_SERVER_URL}/api/v1/stream/stop;
on_update ${WEB_SERVER_URL}/api/v1/stream/heartbeat;
notify_update_timeout 30s;
}
}
}
# Serve the HLS segments over HTTP
http {
server {
listen 8088;
location /hls {
types {
application/vnd.apple.mpegurl m3u8; # proper MIME for HLS playlists
video/mp2t ts; # MPEG-TS segment MIME
}
root /stream;
add_header Cache-Control no-cache; # prevent stale segments
add_header Access-Control-Allow-Origin *; # CORS for browser playback
}
}
}

9
nginx/docker-entrypoint.sh Normal file
View File

@@ -0,0 +1,9 @@
#!/bin/sh
# must only run on certain env files, otherwise it replaces too much
envsubst '${WEB_SERVER_URL}' \
< /etc/nginx/templates/default.conf.template \
> /etc/nginx/nginx.conf
# daemon off required otherwise the container would exit immediately after nginx forks to background.
exec nginx -g "daemon off;"

4
pom.xml
View File

@@ -81,6 +81,10 @@
<artifactId>google-api-client</artifactId>
<version>2.8.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
</dependencies>
<build>

6
src/main/Dockerfile
View File

@@ -15,6 +15,12 @@ RUN apt-get update && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*
ENV POSTGRES_DATA=/app/db
ENV MEDIA_DATA=/app/data/media
ENV STREAM_DATA=/app/data/media/streams
VOLUME ["app/db", "app/data/media/streams", "/app/data/media"]
WORKDIR /app
COPY --from=builder /app/target/*.jar app.jar
EXPOSE 8080

13
src/main/java/com/ddf/vodsystem/configuration/AsyncConfig.java
View File

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

1
src/main/java/com/ddf/vodsystem/configuration/SecurityConfig.java
View File

@@ -35,6 +35,7 @@ public class SecurityConfig {
.requestMatchers("/api/v1/upload", "/api/v1/download/**").permitAll()
.requestMatchers("/api/v1/edit/**", "/api/v1/process/**", "/api/v1/progress/**", "/api/v1/convert/**").permitAll()
.requestMatchers("/api/v1/metadata/**").permitAll()
.requestMatchers("/api/v1/stream/start", "/api/v1/stream/stop", "/api/v1/stream/heartbeat").permitAll()
.anyRequest().authenticated()
)
.sessionManagement(session -> session.sessionCreationPolicy(SessionCreationPolicy.STATELESS)

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

@@ -1,9 +1,11 @@
package com.ddf.vodsystem.controllers;
import com.ddf.vodsystem.dto.APIResponse;
import com.ddf.vodsystem.exceptions.AlreadyStreaming;
import com.ddf.vodsystem.exceptions.FFMPEGException;
import com.ddf.vodsystem.exceptions.JobNotFinished;
import com.ddf.vodsystem.exceptions.JobNotFound;
import com.ddf.vodsystem.exceptions.KeyNotFound;
import com.ddf.vodsystem.exceptions.NotAuthenticated;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
@@ -76,4 +78,18 @@ public class GlobalExceptionHandler {
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.UNAUTHORIZED).body(response);
}
@ExceptionHandler(KeyNotFound.class)
public ResponseEntity<APIResponse<Void>> handleKeyNotFound(KeyNotFound ex) {
logger.error("KeyNotFound: {}", ex.getMessage());
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.NOT_FOUND).body(response);
}
@ExceptionHandler(AlreadyStreaming.class)
public ResponseEntity<APIResponse<Void>> handleAlreadyStreaming(AlreadyStreaming ex) {
logger.error("AlreadyStreaming: {}", ex.getMessage());
APIResponse<Void> response = new APIResponse<>(ERROR, ex.getMessage(), null);
return ResponseEntity.status(HttpStatus.CONFLICT).body(response);
}
}

69
src/main/java/com/ddf/vodsystem/controllers/StreamController.java Normal file
View File

@@ -0,0 +1,69 @@
package com.ddf.vodsystem.controllers;
import com.ddf.vodsystem.controllers.input.ClipSectionRequest;
import com.ddf.vodsystem.dto.APIResponse;
import com.ddf.vodsystem.controllers.input.SaveSectionRequest;
import com.ddf.vodsystem.entities.Stream;
import com.ddf.vodsystem.services.StreamService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.io.IOException;
import java.util.List;
@RestController
@RequestMapping("/api/v1/stream")
public class StreamController {
private static final Logger logger = LoggerFactory.getLogger(StreamController.class);
private final StreamService streamService;
public StreamController(StreamService streamService) {
this.streamService = streamService;
}
// Called by nginx on_publish
@PostMapping("/start")
public ResponseEntity<APIResponse<Stream>> startStream(@RequestParam("name") String streamKey) {
logger.info("Stream start callback for key: {}", streamKey);
Stream stream = streamService.startStream(streamKey);
return ResponseEntity.ok(new APIResponse<>("success", "Stream started", stream));
}
// Called by nginx on_publish_done
@PostMapping("/stop")
public ResponseEntity<APIResponse<Void>> stopStream(@RequestParam("name") String streamKey) {
logger.info("Stream stop callback for key: {}", streamKey);
streamService.endStream(streamKey);
return ResponseEntity.ok(new APIResponse<>("success", "Stream ended", null));
}
// Called by nginx on_update
@PostMapping("/heartbeat")
public ResponseEntity<APIResponse<Void>> heartbeat(@RequestParam("name") String streamKey) {
streamService.heartbeatStream(streamKey);
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.accepted().body(new APIResponse<>("success", "Section saved", null));
}
@PostMapping("/clip")
public ResponseEntity<APIResponse<Void>> saveClip(
@RequestBody ClipSectionRequest request) throws IOException, InterruptedException {
streamService.clip(request.duration());
return ResponseEntity.accepted().body(new APIResponse<>("success", "Clip saved", null));
}
@GetMapping("/history/{userId}")
public ResponseEntity<APIResponse<List<Stream>>> getStreamHistory(@PathVariable Long userId) {
List<Stream> history = streamService.getStreamHistory(userId);
return ResponseEntity.ok(new APIResponse<>("success", "Stream history retrieved", history));
}
}

7
src/main/java/com/ddf/vodsystem/controllers/input/ClipSectionRequest.java Normal file
View File

@@ -0,0 +1,7 @@
package com.ddf.vodsystem.controllers.input;
import jakarta.validation.constraints.NotNull;
public record ClipSectionRequest(
@NotNull float duration
) {}

8
src/main/java/com/ddf/vodsystem/controllers/input/SaveSectionRequest.java Normal file
View File

@@ -0,0 +1,8 @@
package com.ddf.vodsystem.controllers.input;
import java.time.Instant;
import jakarta.validation.constraints.NotNull;
public record SaveSectionRequest(
@NotNull Instant startTime,
@NotNull Instant endTime) {}

31
src/main/java/com/ddf/vodsystem/entities/Stream.java Normal file
View File

@@ -0,0 +1,31 @@
package com.ddf.vodsystem.entities;
import jakarta.persistence.*;
import lombok.Data;
import com.fasterxml.jackson.annotation.JsonIgnore;
import java.time.LocalDateTime;
@Entity
@Table(name = "streams")
@Data
public class Stream {
@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;
@Column(name = "start_time")
private LocalDateTime startDate;
@Column(name = "end_time")
private LocalDateTime endDate;
@Column(name = "last_seen")
private LocalDateTime lastSeen;
}

3
src/main/java/com/ddf/vodsystem/entities/User.java
View File

@@ -34,5 +34,8 @@ public class User {
@Column(name = "created_at")
private LocalDateTime createdAt;
@Column(name = "stream_key")
private String streamKey;
}

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

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

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

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

17
src/main/java/com/ddf/vodsystem/repositories/StreamRepository.java Normal file
View File

@@ -0,0 +1,17 @@
package com.ddf.vodsystem.repositories;
import com.ddf.vodsystem.entities.Stream;
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;
@Repository
public interface StreamRepository extends JpaRepository<Stream, Long> {
Optional<Stream> findByUserAndEndDateIsNull(User user);
List<Stream> findByUser(User user);
List<Stream> findByEndDateIsNullAndLastSeenBefore(LocalDateTime cutoff);
}

3
src/main/java/com/ddf/vodsystem/repositories/UserRepository.java
View File

@@ -11,4 +11,7 @@ import java.util.Optional;
public interface UserRepository extends JpaRepository<User, Long> {
@Query("SELECT u FROM User u WHERE u.googleId = ?1")
Optional<User> findByGoogleId(String googleId);
@Query("SELECT u FROM User u WHERE u.streamKey = ?1")
Optional<User> findByStreamKey(String streamKey);
}

8
src/main/java/com/ddf/vodsystem/services/DirectoryService.java
View File

@@ -59,6 +59,14 @@ public class DirectoryService {
return file;
}
public File getVodFile(String filename) throws IOException {
String dir = outputDir + File.separator + "vods" + File.separator + filename;
File file = new File(dir);
createDirectory(file.getParent());
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");

218
src/main/java/com/ddf/vodsystem/services/StreamService.java Normal file
View File

@@ -0,0 +1,218 @@
package com.ddf.vodsystem.services;
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.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;
@Service
public class StreamService {
private static final Logger logger = LoggerFactory.getLogger(StreamService.class);
private static final int HEARTBEAT_TIMEOUT_SECONDS = 60;
private static final float CLIP_MAX_LENGTH = 180;
private static final int HLS_SEGMENT_LENGTH = 3;
private final StreamRepository streamRepository;
private final UserService userService;
private final StreamActionsService streamActionsService;
private final DirectoryService directoryService;
@Value("${storage.stream}")
private String streamDataPath;
public StreamService(StreamRepository streamRepository,
UserService userService,
StreamActionsService streamActionsService,
DirectoryService directoryService) {
this.streamRepository = streamRepository;
this.userService = userService;
this.streamActionsService = streamActionsService;
this.directoryService = directoryService;
}
public Stream startStream(String streamKey) {
User user = resolveUser(streamKey);
streamRepository.findByUserAndEndDateIsNull(user).ifPresent(existing -> {
throw new AlreadyStreaming("User " + user.getUsername() + " is already streaming.");
});
LocalDateTime now = LocalDateTime.now();
Stream stream = new Stream();
stream.setUser(user);
stream.setStartDate(now);
stream.setLastSeen(now);
return streamRepository.saveAndFlush(stream);
}
public void endStream(String streamKey) {
User user = resolveUser(streamKey);
Stream stream = streamRepository.findByUserAndEndDateIsNull(user)
.orElseThrow(() -> new IllegalStateException(
"No active stream found for user " + user.getUsername()));
stream.setEndDate(LocalDateTime.now());
streamRepository.saveAndFlush(stream);
}
public void heartbeatStream(String streamKey) {
User user = resolveUser(streamKey);
Stream stream = streamRepository.findByUserAndEndDateIsNull(user)
.orElseThrow(() -> new IllegalStateException(
"No active stream found for user " + user.getUsername()));
stream.setLastSeen(LocalDateTime.now());
streamRepository.saveAndFlush(stream);
}
public Optional<Stream> getActiveStream(String streamKey) {
User user = resolveUser(streamKey);
return streamRepository.findByUserAndEndDateIsNull(user);
}
public List<Stream> getStreamHistory(Long userId) {
User streamUser = userService.getUserById(userId)
.orElseThrow(() -> new IllegalArgumentException("User not found: " + userId));
Optional<User> user = userService.getLoggedInUser();
if (user.isEmpty()) {
throw new NotAuthenticated("Log in to see user streams");
}
if (!user.get().equals(streamUser)) {
throw new NotAuthenticated("You are not authenticated to see these user streams");
}
return streamRepository.findByUser(streamUser);
}
/**
* Saves a clip of the currently authenticated user's stream between two points in time.
* <p>
* Locates the recorded {@code .ts} segments overlapping the requested window, computes the
* trim offset into the first (partially-overlapping) segment and the total clip duration,
* then delegates the actual extraction to {@link StreamActionsService#saveSection}.
*
* @param startTime the inclusive start of the section to save; must be before {@code endTime}
* @param endTime the exclusive end of the section to save
* @throws IllegalArgumentException if {@code startTime} is not before {@code endTime},
* or if no stream segments exist in the given range
* @throws NotAuthenticated if no user is currently authenticated
* @throws IOException if reading the stream directory or its segments fails
* @throws InterruptedException if the thread is interrupted while performing the save
*/
public void saveSection(Instant startTime, Instant endTime) throws IOException, InterruptedException {
if (startTime.isAfter(endTime)) {
throw new IllegalArgumentException("Start time must be before end time");
}
User user = userService.getLoggedInUser()
.orElseThrow(() -> new NotAuthenticated("User is not authenticated"));
String streamDirectory = streamDataPath + File.separator + user.getStreamKey();
List<File> fileSegments = getSegmentsInRange(streamDirectory, startTime, endTime);
if (fileSegments.isEmpty()) {
throw new IllegalArgumentException("No stream segments found in the given time range");
}
long firstSegmentMs = parseTimestampMs(fileSegments.getFirst());
float trimOffset = Math.max(0f, (startTime.toEpochMilli() - firstSegmentMs) / 1000f);
float duration = (endTime.toEpochMilli() - startTime.toEpochMilli()) / 1000f;
File outputFile = directoryService.getVodFile(UUID.randomUUID() + ".mp4");
streamActionsService.saveSection(fileSegments, trimOffset, duration, outputFile);
}
/**
* Saves a clip of the last {@code duration} seconds of the currently authenticated user's stream.
* <p>
* Computes a time window ending at the current instant and beginning {@code duration} seconds
* earlier (with millisecond precision), then delegates to {@link #saveSection(Instant, Instant)}.
*
* @param duration the length of the clip in seconds, measured back from now; must be greater
* than {@code 0} and no greater than {@link #CLIP_MAX_LENGTH}. Fractional values
* are supported and resolved to the millisecond.
* @throws IllegalArgumentException if {@code duration} is not in the range {@code (0, CLIP_MAX_LENGTH]},
* or if no stream segments exist in the computed window
* @throws NotAuthenticated if no user is currently authenticated
* @throws IOException if reading the stream directory or its segments fails
* @throws InterruptedException if the thread is interrupted while performing the save
*/
public void clip(float duration) throws IOException, InterruptedException {
if (duration <= 0 || duration > CLIP_MAX_LENGTH) {
throw new IllegalArgumentException("Clip length must be between 0 and " + CLIP_MAX_LENGTH + " seconds");
}
// minusSeconds() is possible, but only does integer seconds, not float
Instant endTime = Instant.now();
Instant startTime = endTime.minus(Duration.ofMillis((long) (duration * 1000)));
saveSection(startTime, endTime);
}
@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));
}
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 + HLS_SEGMENT_LENGTH * 1000) > 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;
}
}

13
src/main/java/com/ddf/vodsystem/services/UserService.java
View File

@@ -16,8 +16,10 @@ import org.springframework.stereotype.Service;
import java.io.IOException;
import java.security.GeneralSecurityException;
import java.security.SecureRandom;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.HexFormat;
import java.util.Optional;
@Service
@@ -67,12 +69,17 @@ public class UserService {
return jwtService.generateToken(user.getId());
}
public Optional<User> getUserByStreamKey(String streamKey) {
return userRepository.findByStreamKey(streamKey);
}
private User createOrUpdateUser(User user) {
Optional<User> existingUser = userRepository.findByGoogleId(user.getGoogleId());
if (existingUser.isEmpty()) {
user.setRole(0);
user.setCreatedAt(LocalDateTime.now());
user.setStreamKey(generateStreamKey());
return userRepository.saveAndFlush(user);
}
@@ -107,4 +114,10 @@ public class UserService {
throw new NotAuthenticated("Invalid ID token: " + e.getMessage());
}
}
private String generateStreamKey() {
byte[] bytes = new byte[24];
new SecureRandom().nextBytes(bytes);
return HexFormat.of().formatHex(bytes);
}
}

54
src/main/java/com/ddf/vodsystem/services/media/StreamActionsService.java Normal file
View File

@@ -0,0 +1,54 @@
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.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(
List<File> segments,
float trimOffset,
float duration,
File outputFile
) throws IOException, InterruptedException {
String concatInput = segments.stream()
.map(File::getAbsolutePath)
.collect(Collectors.joining("|"));
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 with ({} segments) to '{}'",
segments.size(), outputFile.getAbsolutePath());
CommandRunner.run(command, line -> logger.debug("{}", line));
return CompletableFuture.completedFuture(outputFile);
}
}

14
src/main/resources/application-local.properties
View File

@@ -1,13 +1 @@
# Database
spring.datasource.url=jdbc:postgresql://postgres:5432/vodSystem
spring.datasource.username=myuser
spring.datasource.password=mypassword
spring.datasource.driver-class-name=org.postgresql.Driver
spring.jpa.hibernate.ddl-auto=update
spring.jpa.show-sql=true
spring.sql.init.mode=always
spring.sql.init.schema-locations=classpath:db/schema.sql
#spring.sql.init.data-locations=classpath:db/data.sql
# Frontend
frontend.url=http://localhost:5173
spring.docker.compose.file=docker-compose.dev.yml

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

@@ -1,12 +1,21 @@
spring.application.name=vodSystem
spring.profiles.active=local
# VODs
spring.servlet.multipart.max-file-size=2GB
spring.servlet.multipart.max-request-size=2GB
storage.temp.inputs=videos/inputs/
storage.temp.outputs=videos/outputs/
storage.outputs=outputs/
storage.outputs=${MEDIA_DATA}
# Database
spring.datasource.url=${POSTGRES_URL}
spring.datasource.username=${POSTGRES_USER}
spring.datasource.password=${POSTGRES_PASSWORD}
spring.datasource.driver-class-name=org.postgresql.Driver
spring.jpa.hibernate.ddl-auto=update
spring.jpa.show-sql=true
spring.sql.init.mode=always
spring.sql.init.schema-locations=classpath:db/schema.sql
## Security
spring.security.oauth2.client.registration.google.client-id=${GOOGLE_CLIENT_ID}
@@ -16,5 +25,11 @@ spring.security.oauth2.client.registration.google.scope=profile,email
jwt.secret.key=${JWT_SECRET_KEY}
jwt.expiration=3600000
# Frontend
frontend.url=${FRONTEND_URL}
# Stream HLS
storage.stream=${STREAM_DATA}
## Server Configuration
server.servlet.session.timeout=30m

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

@@ -1,5 +1,6 @@
DROP TABLE IF EXISTS clips;
DROP TABLE IF EXISTS users;
-- DROP TABLE IF EXISTS streams;
-- DROP TABLE IF EXISTS clips;
-- DROP TABLE IF EXISTS users;
CREATE TABLE IF NOT EXISTS users (
id BIGSERIAL PRIMARY KEY,
@@ -9,8 +10,19 @@ CREATE TABLE IF NOT EXISTS users (
name VARCHAR(100) NOT NULL,
profile_picture_url VARCHAR(255),
role INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
stream_key VARCHAR(64) NOT NULL UNIQUE
);
CREATE TABLE IF NOT EXISTS streams (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT,
start_time TIMESTAMPTZ DEFAULT NOW(),
end_time TIMESTAMPTZ DEFAULT NULL,
last_seen TIMESTAMPTZ DEFAULT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS clips (
id BIGSERIAL PRIMARY KEY,