diff --git a/src/Cli/Commands/MonitorCommand.php b/src/Cli/Commands/MonitorCommand.php index 1eb5997e..0164becb 100644 --- a/src/Cli/Commands/MonitorCommand.php +++ b/src/Cli/Commands/MonitorCommand.php @@ -11,6 +11,7 @@ use XcVm\Core\Util\StreamUtils; use XcVm\Domain\Server\ServerRepository; use XcVm\Domain\Stream\StreamProcess; use XcVm\Domain\Stream\StreamSorter; +use XcVm\Domain\Stream\StreamSource; use XcVm\Domain\Stream\StreamStateWriter; use XcVm\Streaming\Codec\FFprobeRunner; use XcVm\Streaming\Fanout\FanoutClient; @@ -115,8 +116,7 @@ class MonitorCommand implements CommandInterface { $rCurrentSource = ($rParentID <= 0) ? $rStreamInfo['current_source'] : 'Loopback: #' . $rParentID; $rLastSegment = $rForceSource = null; - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); - $rStreamArguments = $db->get_rows(); + $rStreamArguments = StreamSource::arguments(intval($rStreamID), false, $db); if (0 >= $rStreamInfo['delay_minutes'] && ($rStreamInfo['parent_id'] == 0)) { $rDelay = false; diff --git a/src/Cli/Commands/OndemandCommand.php b/src/Cli/Commands/OndemandCommand.php index ff74b53a..aca45b92 100644 --- a/src/Cli/Commands/OndemandCommand.php +++ b/src/Cli/Commands/OndemandCommand.php @@ -3,11 +3,13 @@ namespace XcVm\Cli\Commands; use XcVm\Cli\CommandInterface; +use XcVm\Core\Cluster\SignalDispatcher; use XcVm\Core\Config\SettingsManager; use XcVm\Core\Config\SettingsRepository; use XcVm\Core\Process\ProcessManager; use XcVm\Domain\Stream\ConnectionTracker; use XcVm\Domain\Stream\StreamProcess; +use XcVm\Domain\Stream\StreamStateWriter; use XcVm\Infrastructure\Redis\RedisManager; use XcVm\Streaming\Fanout\FanoutClient; @@ -163,9 +165,9 @@ class OndemandCommand implements CommandInterface { @unlink($queueFile); @unlink(SIGNALS_TMP_PATH . 'admin_' . $rStreamID); - $db->query("UPDATE streams_servers SET bitrate = NULL, current_source = NULL, to_analyze = 0, pid = NULL, stream_started = NULL, stream_info = NULL, audio_codec = NULL, video_codec = NULL, resolution = NULL, compatible = 0, stream_status = 0, monitor_pid = NULL WHERE stream_id = ? AND server_id = ?", $rStreamID, SERVER_ID); + StreamStateWriter::update(intval($rStreamID), intval(SERVER_ID), ['bitrate' => null, 'current_source' => null, 'to_analyze' => 0, 'pid' => null, 'stream_started' => null, 'stream_info' => null, 'audio_codec' => null, 'video_codec' => null, 'resolution' => null, 'compatible' => 0, 'stream_status' => 0, 'monitor_pid' => null], $db); - $db->query("INSERT INTO signals (server_id, cache, time, custom_data) VALUES (?, 1, ?, ?)", $rMainID, time(), json_encode(['type' => 'update_stream', 'id' => $rStreamID])); + SignalDispatcher::cache(intval($rMainID), ['type' => 'update_stream', 'id' => $rStreamID], false, false, $db); StreamProcess::updateStream($rStreamID); } diff --git a/src/Cli/Commands/ProxyCommand.php b/src/Cli/Commands/ProxyCommand.php index 09f55195..e28776bf 100644 --- a/src/Cli/Commands/ProxyCommand.php +++ b/src/Cli/Commands/ProxyCommand.php @@ -6,6 +6,7 @@ use XcVm\Cli\CommandInterface; use XcVm\Core\Config\SettingsManager; use XcVm\Core\Util\StreamUtils; use XcVm\Domain\Stream\StreamProcess; +use XcVm\Domain\Stream\StreamSource; use XcVm\Domain\Stream\StreamStateWriter; use XcVm\Streaming\Codec\FfmpegPaths; @@ -89,8 +90,7 @@ class ProxyCommand implements CommandInterface { file_put_contents(STREAMS_PATH . $rStreamID . '_.monitor', getmypid()); @unlink(STREAMS_PATH . $rStreamID . '_.pid'); $rStreamInfo = $db->get_row(); - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); - $rStreamArguments = $db->get_rows(true, 'argument_key'); + $rStreamArguments = StreamSource::arguments($rStreamID, true, $db); $this->startProxy($rStreamID, $rStreamInfo, $rStreamArguments, $rFP); diff --git a/src/Cli/Commands/ScannerCommand.php b/src/Cli/Commands/ScannerCommand.php index d3f27df0..ced1da2d 100644 --- a/src/Cli/Commands/ScannerCommand.php +++ b/src/Cli/Commands/ScannerCommand.php @@ -9,6 +9,7 @@ use XcVm\Core\Config\SettingsRepository; use XcVm\Core\Http\CurlClient; use XcVm\Core\Util\StreamUtils; use XcVm\Domain\Stream\StreamSorter; +use XcVm\Domain\Stream\StreamSource; use XcVm\Domain\Stream\StreamStateWriter; use XcVm\Infrastructure\Database\DatabaseAware; use XcVm\Streaming\Codec\FfmpegPaths; @@ -91,8 +92,7 @@ class ScannerCommand implements CommandInterface { foreach ($db->get_rows() as $rRow) { echo '[' . $rRow['id'] . '] - ' . $rRow['stream_display_name'] . "\n"; - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rRow['id']); - $rStreamArguments = $db->get_rows(); + $rStreamArguments = StreamSource::arguments(intval($rRow['id']), false, $db); $rProbesize = (intval($rRow['probesize_ondemand']) ?: 512000); $rAnalyseDuration = '10000000'; $rTimeout = intval($rAnalyseDuration / 1000000) + SettingsManager::get('probe_extra_wait'); diff --git a/src/Cli/CronJobs/CacheEngineCronJob.php b/src/Cli/CronJobs/CacheEngineCronJob.php index 2bdcb88d..bde5aa91 100644 --- a/src/Cli/CronJobs/CacheEngineCronJob.php +++ b/src/Cli/CronJobs/CacheEngineCronJob.php @@ -8,6 +8,7 @@ use XcVm\Core\Config\SettingsManager; use XcVm\Core\Config\SettingsRepository; use XcVm\Core\Process\Multithread; use XcVm\Core\Process\ProcessManager; +use XcVm\Domain\Stream\StreamCacheBuilder; /** * CacheEngineCronJob — cache engine cron job @@ -447,44 +448,24 @@ class CacheEngineCronJob implements CommandInterface { } else { $rMax = $this->rSplit; } - $db->query('SELECT t1.id,t1.epg_id,t1.added,t1.allow_record,t1.year,t1.channel_id,t1.movie_properties,t1.stream_source,t1.tv_archive_server_id,t1.vframes_server_id,t1.tv_archive_duration,t1.stream_icon,t1.custom_sid,t1.category_id,t1.stream_display_name,t1.series_no,t1.direct_source,t1.direct_proxy,t2.type_output,t1.target_container,t2.live,t1.rtmp_output,t1.order,t2.type_key,t1.tmdb_id,t1.adaptive_link FROM `streams` t1 INNER JOIN `streams_types` t2 ON t2.type_id = t1.type LIMIT ' . $rStep . ', ' . $rMax . ';'); + $rRows = StreamCacheBuilder::streamRows($db, null, $rStep, $rMax); } else { - $db->query('SELECT t1.id,t1.epg_id,t1.added,t1.allow_record,t1.year,t1.channel_id,t1.movie_properties,t1.stream_source,t1.tv_archive_server_id,t1.vframes_server_id,t1.tv_archive_duration,t1.stream_icon,t1.custom_sid,t1.category_id,t1.stream_display_name,t1.series_no,t1.direct_source,t1.direct_proxy,t2.type_output,t1.target_container,t2.live,t1.rtmp_output,t1.order,t2.type_key,t1.tmdb_id,t1.adaptive_link FROM `streams` t1 INNER JOIN `streams_types` t2 ON t2.type_id = t1.type WHERE `t1`.`id` IN (' . implode(',', $cacheLockMechanism) . ');'); + $rRows = StreamCacheBuilder::streamRows($db, array_map('intval', $cacheLockMechanism)); } - if ($db->result) { - if ($db->result->rowCount() > 0) { - $rRows = $db->result->fetchAll(\PDO::FETCH_ASSOC); - $rStreamMap = $rStreamIDs = []; - foreach ($rRows as $rRow) { - $rStreamIDs[] = $rRow['id']; - } - if (count($rStreamIDs) > 0) { - if ($db->query('SELECT `stream_id`, `server_id`, `pid`, `to_analyze`, `stream_status`, `monitor_pid`, `on_demand`, `delay_available_at`, `bitrate`, `parent_id`, `on_demand`, `stream_info`, `video_codec`, `audio_codec`, `resolution`, `compatible` FROM `streams_servers` WHERE `stream_id` IN (' . implode(',', $rStreamIDs) . ')')) { - if ($db->result->rowCount() > 0) { - foreach ($db->result->fetchAll(\PDO::FETCH_ASSOC) as $rRow) { - $rStreamMap[intval($rRow['stream_id'])][intval($rRow['server_id'])] = $rRow; - } - } - $db->result = null; - } - } - foreach ($rRows as $rStreamInfo) { - $rExists[] = $rStreamInfo['id']; - if (!$rStreamInfo['direct_source']) { - unset($rStreamInfo['stream_source']); - } - $rOutput = ['info' => $rStreamInfo, 'bouquets' => ($rBouquetMap[intval($rStreamInfo['id'])] ?? []), 'servers' => ($rStreamMap[intval($rStreamInfo['id'])] ?? [])]; - file_put_contents(STREAMS_TMP_PATH . 'stream_' . $rStreamInfo['id'], igbinary_serialize($rOutput)); - } - unset($rRows, $rStreamMap, $rStreamIDs); + if (!empty($rRows)) { + $rStreamMap = StreamCacheBuilder::serverMap($db, array_map(static fn($rRow) => intval($rRow['id']), $rRows)); + foreach ($rRows as $rStreamInfo) { + $rExists[] = $rStreamInfo['id']; + $rID = intval($rStreamInfo['id']); + StreamCacheBuilder::write($rID, StreamCacheBuilder::entry($rStreamInfo, $rBouquetMap[$rID] ?? [], $rStreamMap[$rID] ?? [])); } - $db->result = null; + unset($rRows, $rStreamMap); } } if (count($cacheLockMechanism) > 0) { foreach ($cacheLockMechanism as $rForceID) { - if (!in_array($rForceID, $rExists) && file_exists(STREAMS_TMP_PATH . 'stream_' . $rForceID)) { - unlink(STREAMS_TMP_PATH . 'stream_' . $rForceID); + if (!in_array($rForceID, $rExists)) { + StreamCacheBuilder::remove(intval($rForceID)); } } } diff --git a/src/Domain/Stream/StreamCacheBuilder.php b/src/Domain/Stream/StreamCacheBuilder.php new file mode 100644 index 00000000..b9e16b1f --- /dev/null +++ b/src/Domain/Stream/StreamCacheBuilder.php @@ -0,0 +1,104 @@ +`) that the + * streaming endpoints read instead of the database: the stream's delivery + * fields, its bouquets and its per-server runtime rows. + * + * cron:cache_engine builds it on every node, from MAIN's database, and still + * does. In API mode (Phase 5) a node gets the same entry from MAIN's R2 + * stream delta and `stream_bundle` instead, so the shape lives here, in one + * place, and entry() is pure. + * + * @package XC_VM_Domain_Stream + * @license AGPL-3.0 https://www.gnu.org/licenses/agpl-3.0.html + */ +final class StreamCacheBuilder { + /** `streams` (t1) ⨝ `streams_types` (t2) columns a cache entry carries in `info`. */ + public const STREAM_COLUMNS = 't1.id,t1.epg_id,t1.added,t1.allow_record,t1.year,t1.channel_id,t1.movie_properties,t1.stream_source,t1.tv_archive_server_id,t1.vframes_server_id,t1.tv_archive_duration,t1.stream_icon,t1.custom_sid,t1.category_id,t1.stream_display_name,t1.series_no,t1.direct_source,t1.direct_proxy,t2.type_output,t1.target_container,t2.live,t1.rtmp_output,t1.order,t2.type_key,t1.tmdb_id,t1.adaptive_link'; + + /** `streams_servers` columns a cache entry carries per server. */ + public const SERVER_COLUMNS = ['stream_id', 'server_id', 'pid', 'to_analyze', 'stream_status', 'monitor_pid', 'on_demand', 'delay_available_at', 'bitrate', 'parent_id', 'stream_info', 'video_codec', 'audio_codec', 'resolution', 'compatible']; + + /** + * Stream rows, either a page (offset/limit) or the given ids. + * + * @param list|null $rIDs + * @return list> + */ + public static function streamRows(object $rDb, ?array $rIDs, ?int $rOffset = null, ?int $rLimit = null): array { + $rSql = 'SELECT ' . self::STREAM_COLUMNS . ' FROM `streams` t1 INNER JOIN `streams_types` t2 ON t2.type_id = t1.type'; + if ($rIDs !== null) { + if (empty($rIDs)) { + return []; + } + $rSql .= ' WHERE `t1`.`id` IN (' . implode(',', array_map('intval', $rIDs)) . ');'; + } else { + $rSql .= ' LIMIT ' . intval($rOffset) . ', ' . intval($rLimit) . ';'; + } + if (!$rDb->query($rSql) || !$rDb->result) { + return []; + } + $rRows = $rDb->result->rowCount() > 0 ? $rDb->result->fetchAll(\PDO::FETCH_ASSOC) : []; + $rDb->result = null; + return $rRows; + } + + /** + * Per-server rows of the given streams: stream_id => server_id => row. + * + * @param list $rStreamIDs + * @return array>> + */ + public static function serverMap(object $rDb, array $rStreamIDs): array { + $rMap = []; + if (empty($rStreamIDs)) { + return $rMap; + } + if ($rDb->query('SELECT `' . implode('`, `', self::SERVER_COLUMNS) . '` FROM `streams_servers` WHERE `stream_id` IN (' . implode(',', array_map('intval', $rStreamIDs)) . ')')) { + if ($rDb->result && $rDb->result->rowCount() > 0) { + foreach ($rDb->result->fetchAll(\PDO::FETCH_ASSOC) as $rRow) { + $rMap[intval($rRow['stream_id'])][intval($rRow['server_id'])] = $rRow; + } + } + $rDb->result = null; + } + return $rMap; + } + + /** + * The cache entry for one stream. A stream's source URLs stay out of the + * cache unless it is a direct source (the viewer is sent to the source). + * + * @param array $rStreamRow A streamRows() row. + * @param list $rBouquets Bouquet ids holding the stream. + * @param array> $rServers server_id => serverMap() row. + * @return array{info: array, bouquets: list, servers: array>} + */ + public static function entry(array $rStreamRow, array $rBouquets, array $rServers): array { + if (!$rStreamRow['direct_source']) { + unset($rStreamRow['stream_source']); + } + return ['info' => $rStreamRow, 'bouquets' => $rBouquets, 'servers' => $rServers]; + } + + public static function path(int $rStreamID): string { + return STREAMS_TMP_PATH . 'stream_' . $rStreamID; + } + + /** @param array $rEntry */ + public static function write(int $rStreamID, array $rEntry): void { + file_put_contents(self::path($rStreamID), igbinary_serialize($rEntry)); + } + + /** Drop the entry of a stream that no longer exists. */ + public static function remove(int $rStreamID): void { + if (file_exists(self::path($rStreamID))) { + unlink(self::path($rStreamID)); + } + } +} diff --git a/src/Domain/Stream/StreamProcess.php b/src/Domain/Stream/StreamProcess.php index 268c84f5..3d272abc 100644 --- a/src/Domain/Stream/StreamProcess.php +++ b/src/Domain/Stream/StreamProcess.php @@ -1255,18 +1255,17 @@ class StreamProcess { $rFFMPEGGpu = $rFFMPEG_GPU ?: FfmpegPaths::gpu(); $rFFProbeBin = $rFFPROBE ?: FfmpegPaths::probe(); - $db->query('SELECT * FROM `streams` t1 INNER JOIN `streams_types` t2 ON t2.type_id = t1.type AND t2.live = 1 LEFT JOIN `profiles` t4 ON t1.transcode_profile_id = t4.profile_id WHERE t1.direct_source = 0 AND t1.id = ?', $rStreamID); - if ($db->num_rows() <= 0) { + $rStreamRow = StreamSource::streamRow(intval($rStreamID), true, $db); + if ($rStreamRow === null) { return null; } - $rStream = ['stream_info' => $db->get_row()]; - $db->query('SELECT * FROM `streams_servers` WHERE stream_id = ? AND `server_id` = ?', $rStreamID, SERVER_ID); - if ($db->num_rows() <= 0) { + $rStream = ['stream_info' => $rStreamRow]; + $rServerRow = StreamSource::serverRow(intval($rStreamID), null, $db); + if ($rServerRow === null) { return null; } - $rStream['server_info'] = $db->get_row(); - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); - $rStream['stream_arguments'] = $db->get_rows(); + $rStream['server_info'] = $rServerRow; + $rStream['stream_arguments'] = StreamSource::arguments(intval($rStreamID), false, $db); $rInfo = $rStream['stream_info']; $rParentID = intval($rStream['server_info']['parent_id']); @@ -1966,14 +1965,13 @@ class StreamProcess { $db = self::db(); $rStream = []; $rLoopback = false; - $db->query('SELECT * FROM `streams` t1 INNER JOIN `streams_types` t2 ON t2.type_id = t1.type AND t2.live = 0 LEFT JOIN `profiles` t4 ON t1.transcode_profile_id = t4.profile_id WHERE t1.direct_source = 0 AND t1.id = ?', $rStreamID); - if ($db->num_rows() > 0) { - $rStream['stream_info'] = $db->get_row(); - $db->query('SELECT * FROM `streams_servers` WHERE stream_id = ? AND `server_id` = ?', $rStreamID, SERVER_ID); - if ($db->num_rows() > 0) { - $rStream['server_info'] = $db->get_row(); - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); - $rStream['stream_arguments'] = $db->get_rows(); + $rStreamRow = StreamSource::streamRow(intval($rStreamID), false, $db); + if ($rStreamRow !== null) { + $rStream['stream_info'] = $rStreamRow; + $rServerRow = StreamSource::serverRow(intval($rStreamID), null, $db); + if ($rServerRow !== null) { + $rStream['server_info'] = $rServerRow; + $rStream['stream_arguments'] = StreamSource::arguments(intval($rStreamID), false, $db); list($rStreamSource) = json_decode($rStream['stream_info']['stream_source'], true); if (substr($rStreamSource, 0, 2) == 's:') { @@ -2131,16 +2129,15 @@ class StreamProcess { @unlink(STREAMS_PATH . $rStreamID . '_.pid'); $rStream = []; - $db->query('SELECT * FROM `streams` t1 INNER JOIN `streams_types` t2 ON t2.type_id = t1.type AND t2.live = 1 LEFT JOIN `profiles` t4 ON t1.transcode_profile_id = t4.profile_id WHERE t1.direct_source = 0 AND t1.id = ?', $rStreamID); + $rStreamRow = StreamSource::streamRow(intval($rStreamID), true, $db); - if ($db->num_rows() > 0) { - $rStream['stream_info'] = $db->get_row(); - $db->query('SELECT * FROM `streams_servers` WHERE stream_id = ? AND `server_id` = ?', $rStreamID, SERVER_ID); + if ($rStreamRow !== null) { + $rStream['stream_info'] = $rStreamRow; + $rServerRow = StreamSource::serverRow(intval($rStreamID), null, $db); - if ($db->num_rows() > 0) { - $rStream['server_info'] = $db->get_row(); - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); - $rStream['stream_arguments'] = $db->get_rows(); + if ($rServerRow !== null) { + $rStream['server_info'] = $rServerRow; + $rStream['stream_arguments'] = StreamSource::arguments(intval($rStreamID), false, $db); list($rProbesize, $rAnalyseDuration, $rTimeout) = self::resolveProbeSettings($rStream['server_info']['on_demand'], $rStream['stream_info']['probesize_ondemand'], $rLLOD, $rSettings); $rFFProbee = 'timeout ' . $rTimeout . ' ' . $rFFPROBE . ' {FETCH_OPTIONS} -probesize ' . $rProbesize . ' -analyzeduration ' . $rAnalyseDuration . ' {CONCAT} -i {STREAM_SOURCE} -v quiet -print_format json -show_streams -show_format'; diff --git a/src/Domain/Stream/StreamRowMerge.php b/src/Domain/Stream/StreamRowMerge.php new file mode 100644 index 00000000..6849aaf6 --- /dev/null +++ b/src/Domain/Stream/StreamRowMerge.php @@ -0,0 +1,79 @@ + $rFields + */ + public static function mergeNode(int $rServerID, int $rStreamID, array $rFields, ?object $rDb = null): bool { + return self::apply('`stream_id` = ? AND `server_id` = ?', $rFields, [$rStreamID, $rServerID], $rDb); + } + + /** + * Merge state into a row addressed by its `server_stream_id` (local callers + * that hold the row id; the API path always uses mergeNode()). + * + * @param array $rFields + */ + public static function mergeRow(int $rServerStreamID, array $rFields, ?object $rDb = null): bool { + return self::apply('`server_stream_id` = ?', $rFields, [$rServerStreamID], $rDb); + } + + /** + * The fields a `stream.state` event may carry: runtime-state columns only, + * with credentials stripped from the source URL. Unknown keys are dropped + * rather than refused, because the sender is a remote node. + * + * @param array $rFields + * @return array + */ + public static function eventFields(array $rFields): array { + $rFields = array_intersect_key($rFields, array_flip(StreamStateWriter::STATE_FIELDS)); + if (isset($rFields['current_source']) && is_string($rFields['current_source'])) { + $rFields['current_source'] = Redactor::redact($rFields['current_source']); + } + return $rFields; + } + + /** + * The UPDATE itself. + * + * @param array $rFields + * @param list $rWhereValues + */ + public static function apply(string $rWhere, array $rFields, array $rWhereValues, ?object $rDb = null): bool { + if (empty($rFields)) { + return true; + } + $rUnknown = array_diff(array_keys($rFields), StreamStateWriter::STATE_FIELDS); + if (!empty($rUnknown)) { + throw new \InvalidArgumentException('Not stream runtime state: ' . implode(', ', $rUnknown)); + } + $rSet = []; + foreach (array_keys($rFields) as $rColumn) { + $rSet[] = '`' . $rColumn . '` = ?'; + } + $rDb ??= DatabaseFactory::get(); + return (bool) $rDb->query('UPDATE `streams_servers` SET ' . implode(', ', $rSet) . ' WHERE ' . $rWhere, ...array_values($rFields), ...$rWhereValues); + } +} diff --git a/src/Domain/Stream/StreamSource.php b/src/Domain/Stream/StreamSource.php new file mode 100644 index 00000000..359d3330 --- /dev/null +++ b/src/Domain/Stream/StreamSource.php @@ -0,0 +1,92 @@ +): mixed)|null */ + private static $rLoader = null; + + /** + * The stream's row joined with its type (live or not) and transcode + * profile. Direct-source streams are not run by a node, so they are + * excluded. + * + * @return array|null + */ + public static function streamRow(int $rStreamID, bool $rLive, ?object $rDb = null): ?array { + if (self::$rLoader !== null) { + return (self::$rLoader)('stream', $rStreamID, ['live' => $rLive]); + } + $rDb ??= DatabaseFactory::get(); + $rDb->query('SELECT * FROM `streams` t1 INNER JOIN `streams_types` t2 ON t2.type_id = t1.type AND t2.live = ' . ($rLive ? 1 : 0) . ' LEFT JOIN `profiles` t4 ON t1.transcode_profile_id = t4.profile_id WHERE t1.direct_source = 0 AND t1.id = ?', $rStreamID); + return $rDb->num_rows() > 0 ? $rDb->get_row() : null; + } + + /** + * A node's `streams_servers` row for the stream (this node by default). + * + * @return array|null + */ + public static function serverRow(int $rStreamID, ?int $rServerID = null, ?object $rDb = null): ?array { + $rServerID ??= intval(SERVER_ID); + if (self::$rLoader !== null) { + return (self::$rLoader)('server', $rStreamID, ['server_id' => $rServerID]); + } + $rDb ??= DatabaseFactory::get(); + $rDb->query('SELECT * FROM `streams_servers` WHERE stream_id = ? AND `server_id` = ?', $rStreamID, $rServerID); + return $rDb->num_rows() > 0 ? $rDb->get_row() : null; + } + + /** + * The stream's options joined with their argument definitions, as a list + * or keyed by `argument_key`. + * + * @return array> + */ + public static function arguments(int $rStreamID, bool $rKeyed = false, ?object $rDb = null): array { + if (self::$rLoader !== null) { + return (self::$rLoader)('arguments', $rStreamID, ['keyed' => $rKeyed]); + } + $rDb ??= DatabaseFactory::get(); + $rDb->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); + return ($rKeyed ? $rDb->get_rows(true, 'argument_key') : $rDb->get_rows()) ?: []; + } + + /** + * Just the source list of a stream (`stream_source`, JSON), for callers + * that only pull the source (the fanout proxy hand-off). + * + * @return array{stream_source?: string} + */ + public static function sourceRow(int $rStreamID, ?object $rDb = null): array { + if (self::$rLoader !== null) { + return (self::$rLoader)('source', $rStreamID, []); + } + $rDb ??= DatabaseFactory::get(); + $rDb->query('SELECT `stream_source` FROM `streams` WHERE `id` = ?', $rStreamID); + return $rDb->num_rows() > 0 ? $rDb->get_row() : []; + } + + /** Replace the backend (tests; later the cluster API). Null restores the SQL backend. */ + public static function useLoader(?callable $rLoader): void { + self::$rLoader = $rLoader; + } +} diff --git a/src/Domain/Stream/StreamStateWriter.php b/src/Domain/Stream/StreamStateWriter.php index a12b2ce4..4178aa5a 100644 --- a/src/Domain/Stream/StreamStateWriter.php +++ b/src/Domain/Stream/StreamStateWriter.php @@ -2,8 +2,6 @@ namespace XcVm\Domain\Stream; -use XcVm\Infrastructure\Database\DatabaseFactory; - /** * Stream State Writer * @@ -14,7 +12,8 @@ use XcVm\Infrastructure\Database\DatabaseFactory; * pass the fields to update() or updateRow(), which refuse any column outside * STATE_FIELDS, and the writer applies them through a sink. * - * Today the only sink writes the row in MAIN's database, as before. The + * Today the only sink writes the row in MAIN's database through + * StreamRowMerge, as before. The * cluster API plan (Phase 5) adds one that sends a `stream.state` event * instead, which MAIN merges into the row itself. * @@ -82,11 +81,7 @@ final class StreamStateWriter { if (self::$rSink !== null) { return (bool) (self::$rSink)($rWhere, $rFields, $rWhereValues, $rDb); } - $rSet = []; - foreach (array_keys($rFields) as $rColumn) { - $rSet[] = '`' . $rColumn . '` = ?'; - } - $rDb ??= DatabaseFactory::get(); - return (bool) $rDb->query('UPDATE `streams_servers` SET ' . implode(', ', $rSet) . ' WHERE ' . $rWhere, ...array_values($rFields), ...$rWhereValues); + // Legacy backend: merge into the row in MAIN's database directly. + return StreamRowMerge::apply($rWhere, $rFields, $rWhereValues, $rDb); } } diff --git a/src/Public/stream/live.php b/src/Public/stream/live.php index 95f868f8..1adb25fe 100644 --- a/src/Public/stream/live.php +++ b/src/Public/stream/live.php @@ -6,14 +6,15 @@ use XcVm\Core\Process\ProcessManager; use XcVm\Core\Util\NetworkUtils; use XcVm\Domain\Stream\ConnectionTracker; use XcVm\Domain\Stream\StreamProcess; +use XcVm\Domain\Stream\StreamSource; use XcVm\Infrastructure\Cache\CacheReader; use XcVm\Infrastructure\Database\DatabaseFactory; use XcVm\Infrastructure\Redis\RedisManager; use XcVm\Streaming\AsyncFileOperations; use XcVm\Streaming\Auth\StreamAuth; use XcVm\Streaming\Auth\StreamAuthMiddleware; -use XcVm\Streaming\Delivery\HLSGenerator; use XcVm\Streaming\Codec\FfmpegPaths; +use XcVm\Streaming\Delivery\HLSGenerator; use XcVm\Streaming\Delivery\OffAirHandler; use XcVm\Streaming\Delivery\SegmentReader; use XcVm\Streaming\Delivery\SignalSender; @@ -129,10 +130,8 @@ if ($rChannelInfo) { $rFanout = false; if (!empty($rChannelInfo["proxy"]) && LicenseGate::fanoutUsable()) { DatabaseFactory::connectLazy(); - $db->query('SELECT `stream_source` FROM `streams` WHERE `id` = ?', $rStreamID); - $rStreamRow = ($db->num_rows() > 0 ? $db->get_row() : []); - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); - $rStreamArguments = $db->get_rows(true, 'argument_key'); + $rStreamRow = StreamSource::sourceRow(intval($rStreamID), $db); + $rStreamArguments = StreamSource::arguments(intval($rStreamID), true, $db); $rSource = FanoutClient::buildSource($rStreamRow, $rStreamArguments); $rFanout = !empty($rSource["urls"]) && FanoutClient::register($rStreamID, $rSource); diff --git a/tests/Unit/StreamCacheBuilderSourceTest.php b/tests/Unit/StreamCacheBuilderSourceTest.php new file mode 100644 index 00000000..784d6ba2 --- /dev/null +++ b/tests/Unit/StreamCacheBuilderSourceTest.php @@ -0,0 +1,92 @@ +}> */ + public array $rQueries = []; + public function __construct(private array $rRows) { + } + public function query(string $rSql, mixed ...$rParams): bool { + $this->rQueries[] = [$rSql, $rParams]; + return true; + } + public function num_rows(): int { + return count($this->rRows); + } + public function get_row(): array { + return $this->rRows[0]; + } + public function get_rows(bool $rKeyed = false, string $rKey = ''): array { + return $rKeyed ? array_column($this->rRows, null, $rKey) : $this->rRows; + } + }; + } + + public function testEntryHidesTheSourceUnlessDirect(): void { + $rServers = [2 => ['stream_id' => 9, 'server_id' => 2, 'pid' => 5]]; + $rEntry = StreamCacheBuilder::entry(['id' => 9, 'direct_source' => 0, 'stream_source' => '["http://u:p@src"]'], [3, 4], $rServers); + $this->assertSame(['info' => ['id' => 9, 'direct_source' => 0], 'bouquets' => [3, 4], 'servers' => $rServers], $rEntry); + + $rDirect = StreamCacheBuilder::entry(['id' => 9, 'direct_source' => 1, 'stream_source' => '["http://src"]'], [], []); + $this->assertSame('["http://src"]', $rDirect['info']['stream_source']); + } + + public function testServerColumnsAreTheRuntimeViewOnce(): void { + $this->assertSame(array_values(array_unique(StreamCacheBuilder::SERVER_COLUMNS)), StreamCacheBuilder::SERVER_COLUMNS); + $this->assertContains('parent_id', StreamCacheBuilder::SERVER_COLUMNS); + $this->assertNotContains('cc_info', StreamCacheBuilder::SERVER_COLUMNS); + } + + public function testStreamSourceQueries(): void { + if (!defined('SERVER_ID')) { + define('SERVER_ID', 1); + } + $rDb = $this->recorder([['id' => 4, 'argument_key' => 'user_agent', 'value' => 'UA']]); + $this->assertSame(['id' => 4, 'argument_key' => 'user_agent', 'value' => 'UA'], StreamSource::streamRow(4, true, $rDb)); + $this->assertStringContainsString('AND t2.live = 1 ', $rDb->rQueries[0][0]); + StreamSource::streamRow(4, false, $rDb); + $this->assertStringContainsString('AND t2.live = 0 ', $rDb->rQueries[1][0]); + StreamSource::serverRow(4, 7, $rDb); + $this->assertSame([4, 7], $rDb->rQueries[2][1]); + $this->assertSame(['user_agent' => ['id' => 4, 'argument_key' => 'user_agent', 'value' => 'UA']], StreamSource::arguments(4, true, $rDb)); + + $rEmpty = $this->recorder([]); + $this->assertNull(StreamSource::streamRow(4, true, $rEmpty)); + $this->assertNull(StreamSource::serverRow(4, 7, $rEmpty)); + $this->assertSame([], StreamSource::sourceRow(4, $rEmpty)); + } + + public function testLoaderReplacesTheDatabase(): void { + $rCalls = []; + StreamSource::useLoader(function (string $rWhat, int $rID, array $rOpts) use (&$rCalls) { + $rCalls[] = [$rWhat, $rID, $rOpts]; + return $rWhat === 'arguments' ? [] : ['id' => $rID]; + }); + $this->assertSame(['id' => 3], StreamSource::streamRow(3, true)); + $this->assertSame(['id' => 3], StreamSource::serverRow(3, 2)); + $this->assertSame([], StreamSource::arguments(3, true)); + $this->assertSame([['stream', 3, ['live' => true]], ['server', 3, ['server_id' => 2]], ['arguments', 3, ['keyed' => true]]], $rCalls); + } + + public function testNoNodeSideCodeQueriesStreamOptionsDirectly(): void { + $rRoot = dirname(__DIR__, 2) . '/src/'; + foreach (['Domain/Stream/StreamProcess.php', 'Public/stream/live.php', 'Cli/Commands/ProxyCommand.php', 'Cli/Commands/MonitorCommand.php', 'Cli/Commands/ScannerCommand.php'] as $rFile) { + $this->assertStringNotContainsString('`streams_options` t1', (string) file_get_contents($rRoot . $rFile), $rFile); + } + } +} diff --git a/tests/Unit/StreamRowMergeTest.php b/tests/Unit/StreamRowMergeTest.php new file mode 100644 index 00000000..cae8132c --- /dev/null +++ b/tests/Unit/StreamRowMergeTest.php @@ -0,0 +1,47 @@ +}> */ + public array $rQueries = []; + public function query(string $rSql, mixed ...$rParams): bool { + $this->rQueries[] = [$rSql, $rParams]; + return true; + } + }; + } + + public function testMergeIsKeyedByTheReportingNode(): void { + $rDb = $this->recorder(); + StreamRowMerge::mergeNode(7, 42, ['pid' => 100, 'stream_status' => 0], $rDb); + $this->assertSame([['UPDATE `streams_servers` SET `pid` = ?, `stream_status` = ? WHERE `stream_id` = ? AND `server_id` = ?', [100, 0, 42, 7]]], $rDb->rQueries); + } + + public function testMergeRowAndNothingToMerge(): void { + $rDb = $this->recorder(); + StreamRowMerge::mergeRow(9, ['bitrate' => null], $rDb); + StreamRowMerge::mergeRow(9, [], $rDb); + $this->assertSame([['UPDATE `streams_servers` SET `bitrate` = ? WHERE `server_stream_id` = ?', [null, 9]]], $rDb->rQueries); + } + + public function testDesiredStateIsRefused(): void { + $this->expectException(InvalidArgumentException::class); + StreamRowMerge::mergeNode(1, 1, ['on_demand' => 1], $this->recorder()); + } + + public function testEventFieldsDropForeignKeysAndRedactTheSource(): void { + $this->assertSame( + ['pid' => 5, 'current_source' => 'http://src/live/***/***/1.ts'], + StreamRowMerge::eventFields(['pid' => 5, 'server_id' => 3, 'parent_id' => 2, 'current_source' => 'http://src/live/u/p/1.ts']) + ); + } +}