feat: Implement streaming handlers for live MPEG-TS proxy and VOD HLS transcoding via FFmpeg, and add a release build script.

This commit is contained in:
Michael committed 2025-12-16 07:39:33 +01:00
1 parent c7ba52b88d
commit 8128174ac3
26 files changed
+1529 -927

No files matched your search

+19 -7
View File
@@ -5,13 +5,23 @@ import 'dart:math';
import 'package:shelf/shelf.dart';
import 'package:shelf_router/shelf_router.dart';
import 'package:http/http.dart' as http;
import '../../lib/core/models/playlist_config.dart';
import 'package:xtremflow/core/models/playlist_config.dart';
/// Active FFmpeg processes for VOD transcoding
final Map<String, Process> _vodProcesses = {};
/// Directory for temporary HLS segments
final Directory _hlsTempDir = Directory('/tmp/hls_vod');
final Directory _hlsTempDir = Directory('${Directory.systemTemp.path}/xtremflow_streams');
/// Helper to resolve FFmpeg path (SYSTEM PATH vs Portable)
String _getFFmpegPath() {
if (Platform.isWindows) {
if (File('ffmpeg.exe').existsSync()) return 'ffmpeg.exe';
if (File('bin/ffmpeg.exe').existsSync()) return 'bin/ffmpeg.exe';
if (File('ffmpeg/bin/ffmpeg.exe').existsSync()) return 'ffmpeg/bin/ffmpeg.exe';
}
return 'ffmpeg'; // Default to PATH
}
/// Initialize streaming subsystem
Future<void> initStreaming() async {
@@ -76,11 +86,12 @@ Handler createLiveStreamHandler(Future<PlaylistConfig?> Function(Request) getPla
'-c', 'copy', // Direct stream copy (Very low CPU)
'-f', 'mpegts',
'pipe:1' // Output to stdout
'pipe:1', // Output to stdout
];
try {
final process = await Process.start('ffmpeg', ffmpegArgs);
final ffmpegPath = _getFFmpegPath();
final process = await Process.start(ffmpegPath, ffmpegArgs);
// Cleanup handling: Kill FFmpeg when client disconnects
final controller = StreamController<List<int>>(
@@ -100,7 +111,7 @@ Handler createLiveStreamHandler(Future<PlaylistConfig?> Function(Request) getPla
onError: (e) {
print('[Live] FFmpeg stream error: $e');
controller.addError(e);
}
},
);
// Log errors
@@ -257,10 +268,11 @@ Handler createVodStreamHandler(Future<PlaylistConfig?> Function(Request) getPlay
'-start_number', '0',
// Write playlist to stdout or file? File is easier for static serving
'${streamDir.path}/playlist.m3u8'
'${streamDir.path}/playlist.m3u8',
];
Process.start('ffmpeg', ffmpegArgs).then((process) {
final ffmpegPath = _getFFmpegPath();
Process.start(ffmpegPath, ffmpegArgs).then((process) {
_vodProcesses[streamId] = process;
process.stderr.transform(utf8.decoder).listen((data) {
+41 -20
View File
@@ -12,8 +12,8 @@ class AppDatabase {
/// Initialize database and create tables
Future<void> init() async {
final dbPath = '/app/data/xtremflow.db';
const dbPath = '/app/data/xtremflow.db';
// Ensure data directory exists
final dir = Directory('/app/data');
if (!await dir.exists()) {
@@ -21,7 +21,7 @@ class AppDatabase {
}
_db = sqlite3.open(dbPath);
await _createTables();
print('Database initialized: $dbPath');
}
@@ -79,9 +79,12 @@ class AppDatabase {
''');
// Indexes
_db.execute('CREATE INDEX IF NOT EXISTS idx_sessions_token ON sessions(token)');
_db.execute('CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at)');
_db.execute('CREATE INDEX IF NOT EXISTS idx_playlists_user ON playlists(user_id)');
_db.execute(
'CREATE INDEX IF NOT EXISTS idx_sessions_token ON sessions(token)');
_db.execute(
'CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at)');
_db.execute(
'CREATE INDEX IF NOT EXISTS idx_playlists_user ON playlists(user_id)');
}
/// Seed default admin user if no users exist
@@ -93,10 +96,13 @@ class AppDatabase {
final adminId = _uuid.v4();
final passwordHash = PasswordHasher.hash('admin');
_db.execute('''
_db.execute(
'''
INSERT INTO users (id, username, password_hash, is_admin)
VALUES (?, ?, ?, 1)
''', [adminId, 'admin', passwordHash],);
''',
[adminId, 'admin', passwordHash],
);
print('Default admin user created (username: admin, password: admin)');
}
@@ -148,10 +154,13 @@ class AppDatabase {
final userId = _uuid.v4();
final passwordHash = PasswordHasher.hash(password);
_db.execute('''
_db.execute(
'''
INSERT INTO users (id, username, password_hash, is_admin)
VALUES (?, ?, ?, ?)
''', [userId, username, passwordHash, isAdmin ? 1 : 0],);
''',
[userId, username, passwordHash, isAdmin ? 1 : 0],
);
return User(
id: userId,
@@ -197,10 +206,13 @@ class AppDatabase {
final token = _uuid.v4();
final expiresAt = DateTime.now().add(duration ?? const Duration(days: 7));
_db.execute('''
_db.execute(
'''
INSERT INTO sessions (id, user_id, token, expires_at)
VALUES (?, ?, ?, ?)
''', [sessionId, userId, token, expiresAt.toIso8601String()],);
''',
[sessionId, userId, token, expiresAt.toIso8601String()],
);
return models.Session(
id: sessionId,
@@ -221,7 +233,7 @@ class AppDatabase {
if (result.isEmpty) return null;
final session = models.Session.fromMap(result.first);
// Check if expired
if (session.isExpired) {
deleteSession(token);
@@ -279,10 +291,13 @@ class AppDatabase {
final playlistId = _uuid.v4();
final now = DateTime.now().toIso8601String();
_db.execute('''
_db.execute(
'''
INSERT INTO playlists (id, user_id, name, server_url, username, password, dns, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', [playlistId, userId, name, serverUrl, username, password, dns, now, now],);
''',
[playlistId, userId, name, serverUrl, username, password, dns, now, now],
);
return Playlist(
id: playlistId,
@@ -308,11 +323,14 @@ class AppDatabase {
}) {
final now = DateTime.now().toIso8601String();
_db.execute('''
_db.execute(
'''
UPDATE playlists
SET name = ?, server_url = ?, username = ?, password = ?, dns = ?, updated_at = ?
WHERE id = ?
''', [name, serverUrl, username, password, dns, now, playlistId],);
''',
[name, serverUrl, username, password, dns, now, playlistId],
);
return getPlaylistById(playlistId)!;
}
@@ -338,15 +356,18 @@ class AppDatabase {
/// Update user settings
void updateUserSettings(String userId, String settingsJson) {
final now = DateTime.now().toIso8601String();
// UPSERT (Insert or Replace)
_db.execute('''
_db.execute(
'''
INSERT INTO user_settings (user_id, settings_json, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(user_id) DO UPDATE SET
settings_json = excluded.settings_json,
updated_at = excluded.updated_at
''', [userId, settingsJson, now]);
''',
[userId, settingsJson, now],
);
}
/// Close database connection
+17 -17
View File
@@ -10,7 +10,7 @@ import 'package:args/args.dart';
import 'database/database.dart';
import 'models/user.dart';
import 'models/playlist.dart';
import '../lib/core/models/playlist_config.dart';
import 'package:xtremflow/core/models/playlist_config.dart';
import 'api/auth_handler.dart';
import 'api/users_handler.dart';
import 'api/playlists_handler.dart';
@@ -49,17 +49,17 @@ void main(List<String> args) async {
// Auth endpoints
..mount('/api/auth', authHandler.router)
// Playlists endpoints
..mount('/api/playlists', Pipeline()
..mount('/api/playlists', const Pipeline()
.addMiddleware(authMiddleware(db))
.addHandler(playlistsHandler.router.call))
.addHandler(playlistsHandler.router.call),)
// Users endpoints
..mount('/api/users', Pipeline()
..mount('/api/users', const Pipeline()
.addMiddleware(authMiddleware(db))
.addHandler(usersHandler.router.call))
.addHandler(usersHandler.router.call),)
// Settings endpoints
..mount('/api/settings', Pipeline()
..mount('/api/settings', const Pipeline()
.addMiddleware(authMiddleware(db))
.addHandler(settingsHandler.router.call));
.addHandler(settingsHandler.router.call),);
// Initialize Cleanup Service
final cleanupService = CleanupService();
@@ -70,7 +70,7 @@ void main(List<String> args) async {
cleanupService.start();
// Admin Routes (protected)
apiRouter.mount('/api/admin', Pipeline()
apiRouter.mount('/api/admin', const Pipeline()
.addMiddleware(authMiddleware(db))
.addHandler((Request request) {
final router = Router();
@@ -90,7 +90,7 @@ void main(List<String> args) async {
});
return router(request);
}));
}),);
// Create static handler
final baseStaticHandler = createStaticHandler(
@@ -100,7 +100,7 @@ void main(List<String> args) async {
);
// Wrap static handler to enforce cache policies
Handler staticHandler = (Request request) async {
FutureOr<Response> staticHandler(Request request) async {
final response = await baseStaticHandler(request);
// Disable cache for entry points to ensure updates are seen immediately
@@ -113,17 +113,17 @@ void main(List<String> args) async {
'Cache-Control': 'no-store, no-cache, must-revalidate, max-age=0',
'Pragma': 'no-cache',
'Expires': '0',
});
},);
}
// Allow aggressive caching for hashed assets
return response.change(headers: {
'Cache-Control': 'public, max-age=86400',
});
};
},);
}
// Helper to get playlist from request
Future<PlaylistConfig?> _getPlaylist(Request request) async {
Future<PlaylistConfig?> getPlaylist(Request request) async {
Playlist? playlist;
final user = request.context['user'] as User?;
@@ -155,8 +155,8 @@ void main(List<String> args) async {
// Create Streaming Router (Mount handlers on correct paths)
final streamingRouter = Router()
..mount('/api/live', createLiveStreamHandler(_getPlaylist))
..mount('/api/vod', createVodStreamHandler(_getPlaylist));
..mount('/api/live', createLiveStreamHandler(getPlaylist))
..mount('/api/vod', createVodStreamHandler(getPlaylist));
// Main handler with API proxy and NEW Streaming Handlers
final handler = Cascade()
@@ -167,7 +167,7 @@ void main(List<String> args) async {
.handler;
// Add middleware
final pipeline = Pipeline()
final pipeline = const Pipeline()
.addMiddleware(logRequests())
.addMiddleware(securityHeadersMiddleware())
.addMiddleware(honeypotMiddleware())