doomtube/beeta
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial commit - realms platform

Browse source
This commit is contained in:
doomtube 2026-01-05 22:54:27 -05:00
parent c590ab6d18
commit c717c3751c
234 changed files with 74103 additions and 15231 deletions
Download patch file Download diff file Expand all files Collapse all files

321
backend/src/controllers/StreamController.cpp
View file

@ -4,6 +4,9 @@
#include "../services/RedisHelper.h"
#include "../services/OmeClient.h"
#include "../services/AuthService.h"
#include "../services/RestreamService.h"
#include "../common/HttpHelpers.h"
#include "../common/AuthHelpers.h"
#include <drogon/utils/Utilities.h>
#include <drogon/Cookie.h>
#include <random>
@ -15,24 +18,10 @@ using namespace drogon::orm;
// Helper functions at the top
namespace {
// JSON response helper - saves 6-8 lines per endpoint
HttpResponsePtr jsonResp(const Json::Value& j, HttpStatusCode c = k200OK) {
auto r = HttpResponse::newHttpJsonResponse(j);
r->setStatusCode(c);
return r;
}
HttpResponsePtr jsonOk(const Json::Value& data) {
return jsonResp(data);
}
HttpResponsePtr jsonError(const std::string& error, HttpStatusCode code = k400BadRequest) {
Json::Value j;
j["success"] = false;
j["error"] = error;
return jsonResp(j, code);
}
// Quick JSON builder for common patterns
Json::Value json(std::initializer_list<std::pair<const char*, Json::Value>> items) {
Json::Value j;
@ -41,19 +30,6 @@ namespace {
}
return j;
}
UserInfo getUserFromRequest(const HttpRequestPtr &req) {
UserInfo user;
std::string auth = req->getHeader("Authorization");
if (auth.empty() || auth.substr(0, 7) != "Bearer ") {
return user;
}
std::string token = auth.substr(7);
AuthService::getInstance().validateToken(token, user);
return user;
}
}
// Static member definitions
@ -122,10 +98,7 @@ void StreamController::disconnectStream(const HttpRequestPtr &req,
}
});
}
>> [callback](const DrogonDbException& e) {
LOG_ERROR << "Database error: " << e.base().what();
callback(jsonError("Database error"));
};
>> DB_ERROR(callback, "disconnect stream");
}
void StreamController::getStreamStats(const HttpRequestPtr &,
@ -252,7 +225,8 @@ void StreamController::heartbeat(const HttpRequestPtr &req,
return;
}
services::RedisHelper::instance().expireAsync("viewer_token:" + token, 30,
// Refresh token TTL to 5 minutes on heartbeat
services::RedisHelper::instance().expireAsync("viewer_token:" + token, 300,
[callback](bool success) {
if (!success) {
callback(jsonResp({}, k500InternalServerError));
@ -285,29 +259,29 @@ void StreamWebSocketController::handleNewMessage(const WebSocketConnectionPtr&,
void StreamWebSocketController::handleNewConnection(const HttpRequestPtr &req,
const WebSocketConnectionPtr& wsConnPtr) {
LOG_INFO << "New WebSocket connection established";
// Allow anonymous connections for receiving public broadcasts (stream_live/stream_offline)
// These are used by the home page to get instant updates
std::lock_guard<std::mutex> lock(connectionsMutex_);
connections_.insert(wsConnPtr);
auto token = req->getCookie("viewer_token");
if (token.empty()) {
LOG_WARN << "WebSocket connection without viewer token";
wsConnPtr->shutdown();
return;
}
RedisHelper::getKeyAsync("viewer_token:" + token,
[wsConnPtr, token](const std::string& streamKey) {
if (streamKey.empty()) {
LOG_WARN << "Invalid viewer token";
wsConnPtr->shutdown();
return;
if (!token.empty()) {
// If viewer token is provided, validate and track it
RedisHelper::getKeyAsync("viewer_token:" + token,
[wsConnPtr, token](const std::string& streamKey) {
if (!streamKey.empty()) {
std::lock_guard<std::mutex> lock(connectionsMutex_);
tokenConnections_[token].insert(wsConnPtr);
LOG_INFO << "WebSocket authenticated for stream: " << streamKey;
} else {
LOG_DEBUG << "WebSocket with invalid/expired viewer token - treating as anonymous";
}
}
std::lock_guard<std::mutex> lock(connectionsMutex_);
tokenConnections_[token].insert(wsConnPtr);
connections_.insert(wsConnPtr);
LOG_INFO << "WebSocket authenticated for stream: " << streamKey;
}
);
);
} else {
LOG_DEBUG << "Anonymous WebSocket connection (no viewer token)";
}
}
void StreamWebSocketController::handleConnectionClosed(const WebSocketConnectionPtr& wsConnPtr) {
@ -360,11 +334,248 @@ void StreamWebSocketController::broadcastKeyUpdate(const std::string& userId, co
void StreamWebSocketController::broadcastStatsUpdate(const Json::Value& stats) {
std::string jsonStr = Json::FastWriter().write(stats);
std::lock_guard<std::mutex> lock(connectionsMutex_);
for (const auto& conn : connections_) {
if (conn->connected()) {
conn->send(jsonStr);
}
}
}
// OvenMediaEngine Webhook Handlers
void StreamController::handleOmeWebhook(const HttpRequestPtr &req,
std::function<void(const HttpResponsePtr &)> &&callback) {
auto jsonPtr = req->getJsonObject();
if (!jsonPtr) {
LOG_WARN << "OME webhook received with invalid JSON";
callback(jsonError("Invalid JSON", k400BadRequest));
return;
}
const auto& payload = *jsonPtr;
std::string eventType = payload.get("eventType", "").asString();
LOG_INFO << "OME Webhook received: " << eventType;
LOG_DEBUG << "OME Webhook payload: " << payload.toStyledString();
// Extract stream information
std::string streamName;
if (payload.isMember("stream") && payload["stream"].isMember("name")) {
streamName = payload["stream"]["name"].asString();
} else if (payload.isMember("streamName")) {
streamName = payload["streamName"].asString();
}
if (streamName.empty()) {
LOG_WARN << "OME webhook missing stream name";
callback(jsonOk(json({{"success", true}, {"message", "Acknowledged"}})));
return;
}
auto dbClient = app().getDbClient();
if (eventType == "streamCreated" || eventType == "stream.created" || eventType == "publish") {
// Stream started - mark realm as live immediately
LOG_INFO << "Stream started via webhook: " << streamName;
*dbClient << "UPDATE realms SET is_live = true, viewer_count = 0, "
"updated_at = CURRENT_TIMESTAMP WHERE stream_key = $1 RETURNING id"
<< streamName
>> [streamName](const Result& r) {
LOG_INFO << "Realm marked as live via webhook: " << streamName;
// Broadcast to WebSocket clients
Json::Value msg;
msg["type"] = "stream_live";
msg["stream_key"] = streamName;
msg["is_live"] = true;
StreamWebSocketController::broadcastStatsUpdate(msg);
// Trigger immediate stats fetch
StatsService::getInstance().updateStreamStats(streamName);
// Pre-warm thumbnail cache so it's ready when users see the stream
// This makes an async request to generate the thumbnail in the background
auto client = HttpClient::newHttpClient("http://localhost:8088");
auto req = HttpRequest::newHttpRequest();
req->setPath("/thumb/" + streamName + ".webp");
req->setMethod(drogon::Get);
client->sendRequest(req, [streamName](ReqResult result, const HttpResponsePtr& response) {
if (result == ReqResult::Ok && response && response->statusCode() == k200OK) {
LOG_INFO << "Thumbnail pre-warmed for stream: " << streamName;
} else {
LOG_DEBUG << "Thumbnail pre-warm pending for: " << streamName << " (stream may still be initializing)";
}
}, 10.0); // 10 second timeout for thumbnail generation
// Start restream destinations if realm has any
if (!r.empty()) {
int64_t realmId = r[0]["id"].as<int64_t>();
RestreamService::getInstance().startAllDestinations(streamName, realmId);
}
}
>> [streamName](const DrogonDbException& e) {
LOG_ERROR << "Failed to mark realm live via webhook: " << e.base().what();
};
}
else if (eventType == "streamDeleted" || eventType == "stream.deleted" || eventType == "unpublish") {
// Stream ended - mark realm as offline immediately
LOG_INFO << "Stream ended via webhook: " << streamName;
*dbClient << "UPDATE realms SET is_live = false, viewer_count = 0, "
"updated_at = CURRENT_TIMESTAMP WHERE stream_key = $1 RETURNING id"
<< streamName
>> [streamName](const Result& r) {
LOG_INFO << "Realm marked as offline via webhook: " << streamName;
// Broadcast to WebSocket clients
Json::Value msg;
msg["type"] = "stream_offline";
msg["stream_key"] = streamName;
msg["is_live"] = false;
StreamWebSocketController::broadcastStatsUpdate(msg);
// Stop all restream destinations
if (!r.empty()) {
int64_t realmId = r[0]["id"].as<int64_t>();
RestreamService::getInstance().stopAllDestinations(streamName, realmId);
}
}
>> [streamName](const DrogonDbException& e) {
LOG_ERROR << "Failed to mark realm offline via webhook: " << e.base().what();
};
}
else if (eventType == "sessionCreated" || eventType == "viewer.connected") {
// Viewer connected
LOG_INFO << "Viewer connected to stream: " << streamName;
StatsService::getInstance().updateStreamStats(streamName);
}
else if (eventType == "sessionDeleted" || eventType == "viewer.disconnected") {
// Viewer disconnected
LOG_INFO << "Viewer disconnected from stream: " << streamName;
StatsService::getInstance().updateStreamStats(streamName);
}
// Always respond with success to acknowledge the webhook
callback(jsonOk(json({{"success", true}, {"message", "Webhook processed"}})));
}
void StreamController::handleOmeAdmission(const HttpRequestPtr &req,
std::function<void(const HttpResponsePtr &)> &&callback) {
// Admission webhook - validates if a stream is allowed to publish/play
// OME sends: { "client": {...}, "request": { "direction", "protocol", "status", "url", ... } }
auto jsonPtr = req->getJsonObject();
if (!jsonPtr) {
LOG_WARN << "OME admission webhook received with invalid JSON";
callback(jsonError("Invalid JSON", k400BadRequest));
return;
}
const auto& payload = *jsonPtr;
LOG_INFO << "OME Admission webhook: " << payload.toStyledString();
// Check if this is a "closing" status - just acknowledge it
if (payload.isMember("request") && payload["request"].isMember("status")) {
std::string status = payload["request"]["status"].asString();
if (status == "closing") {
LOG_INFO << "OME admission closing notification";
Json::Value response;
callback(jsonOk(response)); // Empty response for closing
return;
}
}
// Extract stream key from URL: rtmp://host:port/app/STREAM_KEY or similar
std::string streamKey;
if (payload.isMember("request") && payload["request"].isMember("url")) {
std::string url = payload["request"]["url"].asString();
// URL format: scheme://host[:port]/app/stream_key[/file][?query]
// Find the stream key after /app/
size_t appPos = url.find("/app/");
if (appPos != std::string::npos) {
std::string afterApp = url.substr(appPos + 5); // Skip "/app/"
// Remove any trailing path or query string
size_t endPos = afterApp.find_first_of("/?");
if (endPos != std::string::npos) {
streamKey = afterApp.substr(0, endPos);
} else {
streamKey = afterApp;
}
}
LOG_INFO << "Extracted stream key from URL: " << streamKey << " (URL: " << url << ")";
}
if (streamKey.empty()) {
LOG_WARN << "OME admission webhook: could not extract stream key, allowing by default";
Json::Value response;
response["allowed"] = true;
callback(jsonOk(response));
return;
}
// Check direction - only validate "incoming" (publish) requests
std::string direction;
if (payload.isMember("request") && payload["request"].isMember("direction")) {
direction = payload["request"]["direction"].asString();
}
if (direction == "outgoing") {
// Playback request - allow all for now (could add viewer auth later)
LOG_INFO << "Allowing outgoing (playback) request for: " << streamKey;
Json::Value response;
response["allowed"] = true;
callback(jsonOk(response));
return;
}
// Validate stream key against database for incoming (publish) requests
auto dbClient = app().getDbClient();
*dbClient << "SELECT id FROM realms WHERE stream_key = $1 AND is_active = true"
<< streamKey
>> [callback, streamKey](const Result& r) {
Json::Value response;
if (!r.empty()) {
LOG_INFO << "Stream key validated for admission: " << streamKey;
response["allowed"] = true;
// Mark stream as live immediately when publishing is approved
int64_t realmId = r[0]["id"].as<int64_t>();
auto db = app().getDbClient();
*db << "UPDATE realms SET is_live = true, viewer_count = 0, "
"updated_at = CURRENT_TIMESTAMP WHERE id = $1"
<< realmId
>> [streamKey, realmId](const Result&) {
LOG_INFO << "Realm marked live on admission: " << streamKey;
// Broadcast to WebSocket clients
Json::Value msg;
msg["type"] = "stream_live";
msg["stream_key"] = streamKey;
msg["is_live"] = true;
StreamWebSocketController::broadcastStatsUpdate(msg);
// Trigger stats fetch
StatsService::getInstance().updateStreamStats(streamKey);
// Start restream destinations
RestreamService::getInstance().startAllDestinations(streamKey, realmId);
}
>> [streamKey](const DrogonDbException& e) {
LOG_ERROR << "Failed to mark realm live on admission: " << e.base().what();
};
} else {
LOG_WARN << "Invalid stream key rejected: " << streamKey;
response["allowed"] = false;
response["reason"] = "Invalid or inactive stream key";
}
callback(jsonOk(response));
}
>> [callback, streamKey](const DrogonDbException& e) {
LOG_ERROR << "Database error during admission check: " << e.base().what();
// Allow on DB error to prevent blocking legitimate streams
Json::Value response;
response["allowed"] = true;
callback(jsonOk(response));
};
}