mirror of
git://f0xx.org/ac/ac-deploy
synced 2026-07-29 09:18:53 +03:00
Enable lab media_pipelines in composed config and wire ac-ms-media-transcode.
Overlay media pipeline PHP/JS/SQL on compose, nginx HLS location, enable feature in lab config.php, and ensure media storage dirs on BE nodes. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Anton Afanasyeu
227
sim/cluster0/lab-seeds/backend/src/MediaPipelineRunner.php
Normal file
227
sim/cluster0/lab-seeds/backend/src/MediaPipelineRunner.php
Normal file
@@ -0,0 +1,227 @@
|
||||
<?php
|
||||
declare(strict_types=1);
|
||||
|
||||
/**
|
||||
* Invokes ac-ms-media-transcode shell scripts on the BE host (lab / dev).
|
||||
*/
|
||||
final class MediaPipelineRunner {
|
||||
public static function scriptRoot(): string {
|
||||
$root = trim((string) cfg('media_pipelines.script_root', ''));
|
||||
if ($root !== '' && is_dir($root)) {
|
||||
return rtrim($root, '/');
|
||||
}
|
||||
$env = getenv('AC_MEDIA_PIPELINES_ROOT');
|
||||
if (is_string($env) && $env !== '' && is_dir($env)) {
|
||||
return rtrim($env, '/');
|
||||
}
|
||||
$fallback = '/var/www/ac/workspace/ac-ms-media-transcode';
|
||||
if (is_dir($fallback)) {
|
||||
return $fallback;
|
||||
}
|
||||
throw new RuntimeException(
|
||||
'ac-ms-media-transcode script root not found — set media_pipelines.script_root in config.php'
|
||||
);
|
||||
}
|
||||
|
||||
public static function logoPath(): string {
|
||||
$logo = trim((string) cfg('media_pipelines.logo_path', ''));
|
||||
if ($logo !== '' && is_file($logo)) {
|
||||
return $logo;
|
||||
}
|
||||
$default = self::scriptRoot() . '/assets/pirate_drift.png';
|
||||
if (is_file($default)) {
|
||||
return $default;
|
||||
}
|
||||
throw new RuntimeException('pirate_drift.png not found under ac-ms-media-transcode/assets/');
|
||||
}
|
||||
|
||||
public static function storageRoot(): string {
|
||||
$root = trim((string) cfg('media_pipelines.storage_root', ''));
|
||||
if ($root === '') {
|
||||
$root = self::scriptRoot() . '/storage';
|
||||
}
|
||||
if (!is_dir($root)) {
|
||||
mkdir($root, 0755, true);
|
||||
}
|
||||
return rtrim($root, '/');
|
||||
}
|
||||
|
||||
public static function hlsFilesystemRoot(): string {
|
||||
$root = trim((string) cfg('media_pipelines.hls_filesystem_root', ''));
|
||||
if ($root === '') {
|
||||
$root = self::storageRoot() . '/hls';
|
||||
}
|
||||
if (!is_dir($root)) {
|
||||
mkdir($root, 0755, true);
|
||||
}
|
||||
return rtrim($root, '/');
|
||||
}
|
||||
|
||||
public static function hlsPublicPrefix(): string {
|
||||
return rtrim((string) cfg('media_pipelines.hls_url_prefix', '/app/androidcast_project/media/hls'), '/');
|
||||
}
|
||||
|
||||
/** @param array<string, mixed> $pipeline */
|
||||
public static function workDir(array $pipeline): string {
|
||||
$base = self::storageRoot() . '/pipelines/' . (int) $pipeline['id'];
|
||||
if (!is_dir($base)) {
|
||||
mkdir($base, 0755, true);
|
||||
}
|
||||
return $base;
|
||||
}
|
||||
|
||||
/** @param array<string, mixed> $pipeline */
|
||||
public static function probe(array $pipeline): array {
|
||||
$root = self::scriptRoot();
|
||||
$cmd = escapeshellcmd($root . '/bin/probe-source.sh') . ' '
|
||||
. escapeshellarg((string) $pipeline['source_url']);
|
||||
$out = [];
|
||||
$code = 0;
|
||||
exec($cmd . ' 2>&1', $out, $code);
|
||||
$text = trim(implode("\n", $out));
|
||||
if ($code !== 0) {
|
||||
return ['ok' => false, 'error' => $text !== '' ? $text : 'probe failed'];
|
||||
}
|
||||
$json = json_decode($text, true);
|
||||
if (!is_array($json)) {
|
||||
return ['ok' => false, 'error' => 'invalid probe json', 'raw' => $text];
|
||||
}
|
||||
return ['ok' => true, 'probe' => $json];
|
||||
}
|
||||
|
||||
/** @param array<string, mixed> $pipeline */
|
||||
public static function start(array $pipeline): array {
|
||||
if (!MediaPipelinesRepository::gateCanOperate()) {
|
||||
return ['ok' => false, 'error' => 'forbidden'];
|
||||
}
|
||||
$id = (int) $pipeline['id'];
|
||||
$work = self::workDir($pipeline);
|
||||
$hlsSlug = (string) ($pipeline['hls_slug'] ?? ('p' . $id));
|
||||
$hlsDir = self::hlsFilesystemRoot() . '/' . $hlsSlug;
|
||||
$sourcesFile = $work . '/sources.list';
|
||||
$sinksFile = $work . '/sinks.list';
|
||||
$localEnv = $work . '/local.env';
|
||||
|
||||
$line = (string) $pipeline['source_url'];
|
||||
if (!empty($pipeline['source_fmt'])) {
|
||||
$line .= ' ' . (string) $pipeline['source_fmt'];
|
||||
}
|
||||
file_put_contents($sourcesFile, $line . "\n");
|
||||
|
||||
$defaultSinks = self::scriptRoot() . '/config/sinks.list';
|
||||
if (is_readable($defaultSinks)) {
|
||||
copy($defaultSinks, $sinksFile);
|
||||
} else {
|
||||
file_put_contents($sinksFile, "# no sinks\n");
|
||||
}
|
||||
|
||||
$envLines = [
|
||||
'SOURCES_LIST=' . $sourcesFile,
|
||||
'SINKS_LIST=' . $sinksFile,
|
||||
'STORAGE_DIR=' . $work,
|
||||
'RUN_DIR=' . $work . '/run',
|
||||
'HLS_OUTPUT_DIR=' . $hlsDir,
|
||||
'LOGO_PATH=' . self::logoPath(),
|
||||
'FILTER_PROFILE=' . (string) ($pipeline['filter_profile'] ?? 'default'),
|
||||
'OUTPUT_PRESET=' . (string) ($pipeline['output_preset'] ?? '1080p'),
|
||||
'YTDLP_ENABLED=' . (cfg('media_pipelines.ytdlp_enabled', true) ? '1' : '0'),
|
||||
];
|
||||
if (!empty($pipeline['audio_source'])) {
|
||||
$envLines[] = 'AUDIO_SOURCE=' . (string) $pipeline['audio_source'];
|
||||
}
|
||||
file_put_contents($localEnv, implode("\n", $envLines) . "\n");
|
||||
|
||||
MediaPipelinesRepository::updateStatus($id, 'starting');
|
||||
|
||||
self::writeWrapper($work, $runMode);
|
||||
|
||||
$cmd = 'nohup ' . escapeshellarg($work . '/invoke.sh')
|
||||
. ' >> ' . escapeshellarg($work . '/runner.log') . ' 2>&1 & echo $!';
|
||||
$pidLine = trim((string) shell_exec($cmd));
|
||||
$pid = (int) $pidLine;
|
||||
|
||||
$hlsUrl = self::hlsPublicPrefix() . '/' . rawurlencode($hlsSlug) . '/index.m3u8';
|
||||
MediaPipelinesRepository::updateRuntime($id, [
|
||||
'status' => 'running',
|
||||
'pid_encode' => $pid > 0 ? $pid : null,
|
||||
'work_dir' => $work,
|
||||
'hls_playlist_url' => $hlsUrl,
|
||||
'last_error' => null,
|
||||
]);
|
||||
|
||||
return [
|
||||
'ok' => true,
|
||||
'pid' => $pid,
|
||||
'hls_url' => $hlsUrl,
|
||||
'work_dir' => $work,
|
||||
];
|
||||
}
|
||||
|
||||
public static function stop(int $pipelineId, int $userId, int $companyId): array {
|
||||
$pipeline = MediaPipelinesRepository::getById($pipelineId, $userId, $companyId);
|
||||
if ($pipeline === null) {
|
||||
return ['ok' => false, 'error' => 'not_found'];
|
||||
}
|
||||
MediaPipelinesRepository::updateStatus($pipelineId, 'stopping');
|
||||
$root = self::scriptRoot();
|
||||
$cmd = 'cd ' . escapeshellarg($root) . ' && '
|
||||
. escapeshellcmd($root . '/bin/stop.sh') . ' 2>/dev/null; true';
|
||||
exec($cmd);
|
||||
foreach (['pid_encode', 'pid_fanout'] as $key) {
|
||||
$pid = (int) ($pipeline[$key] ?? 0);
|
||||
if ($pid > 0) {
|
||||
exec('kill -15 ' . $pid . ' 2>/dev/null; kill -9 ' . $pid . ' 2>/dev/null');
|
||||
}
|
||||
}
|
||||
MediaPipelinesRepository::updateRuntime($pipelineId, [
|
||||
'status' => 'idle',
|
||||
'pid_encode' => null,
|
||||
'pid_fanout' => null,
|
||||
'last_error' => null,
|
||||
]);
|
||||
return ['ok' => true];
|
||||
}
|
||||
|
||||
private static function writeWrapper(string $work, string $runMode): void {
|
||||
$root = self::scriptRoot();
|
||||
$script = $runMode === 'encode_only' ? 'run-encode-only.sh' : 'run.sh';
|
||||
$wrapper = $work . '/invoke.sh';
|
||||
$body = <<<'SH'
|
||||
#!/bin/sh
|
||||
set -eu
|
||||
WORK_DIR="@WORK@"
|
||||
ROOT="@ROOT@"
|
||||
if [ -f "${WORK_DIR}/local.env" ]; then
|
||||
set -a
|
||||
# shellcheck disable=SC1090
|
||||
. "${WORK_DIR}/local.env"
|
||||
set +a
|
||||
fi
|
||||
# shellcheck disable=SC1091
|
||||
. "${ROOT}/lib/common.sh"
|
||||
mkdir -p "${RUN_DIR:-${WORK_DIR}/run}"
|
||||
exec "${ROOT}/bin/@SCRIPT@" 1
|
||||
SH;
|
||||
$body = str_replace(
|
||||
['@WORK@', '@ROOT@', '@SCRIPT@'],
|
||||
[$work, $root, $script],
|
||||
$body
|
||||
);
|
||||
file_put_contents($wrapper, $body);
|
||||
chmod($wrapper, 0755);
|
||||
}
|
||||
|
||||
/** @return list<string> */
|
||||
public static function tailLog(array $pipeline, int $lines = 80): array {
|
||||
$log = ($pipeline['work_dir'] ?? '') . '/encode.log';
|
||||
if (!is_string($log) || !is_readable($log)) {
|
||||
$log = self::storageRoot() . '/encode.log';
|
||||
}
|
||||
if (!is_readable($log)) {
|
||||
return [];
|
||||
}
|
||||
$out = [];
|
||||
exec('tail -n ' . (int) $lines . ' ' . escapeshellarg($log), $out);
|
||||
return $out;
|
||||
}
|
||||
}
|
||||
185
sim/cluster0/lab-seeds/backend/src/MediaPipelinesRepository.php
Normal file
185
sim/cluster0/lab-seeds/backend/src/MediaPipelinesRepository.php
Normal file
@@ -0,0 +1,185 @@
|
||||
<?php
|
||||
declare(strict_types=1);
|
||||
|
||||
/**
|
||||
* Experimental V.2x media pipelines — DB + access gate + CRUD.
|
||||
* Control plane only; workers run streaming_v2 shell scripts on BE.
|
||||
*/
|
||||
final class MediaPipelinesRepository {
|
||||
private static bool $schemaEnsured = false;
|
||||
|
||||
public static function gateAllowed(): bool {
|
||||
if (!cfg('media_pipelines.enabled', false)) {
|
||||
return false;
|
||||
}
|
||||
if (cfg('media_pipelines.developers_only', true)) {
|
||||
return Rbac::isGlobalAdmin()
|
||||
|| Rbac::can('media_pipelines_view')
|
||||
|| Rbac::can('media_pipelines_operate');
|
||||
}
|
||||
return Auth::user() !== null;
|
||||
}
|
||||
|
||||
public static function gateCanOperate(): bool {
|
||||
if (!self::gateAllowed()) {
|
||||
return false;
|
||||
}
|
||||
return Rbac::isGlobalAdmin() || Rbac::can('media_pipelines_operate');
|
||||
}
|
||||
|
||||
public static function ensureSchema(): void {
|
||||
if (self::$schemaEnsured) {
|
||||
return;
|
||||
}
|
||||
$pdo = Database::pdo();
|
||||
Database::withMigrationLock($pdo, 'media_pipelines', static function () use ($pdo): void {
|
||||
if (Database::tableExists($pdo, 'media_pipelines')) {
|
||||
return;
|
||||
}
|
||||
if (Database::isMysql()) {
|
||||
$sql = file_get_contents(__DIR__ . '/../sql/migrations/081_media_pipelines.sql') ?: '';
|
||||
if ($sql !== '') {
|
||||
$pdo->exec($sql);
|
||||
}
|
||||
return;
|
||||
}
|
||||
$sql = file_get_contents(__DIR__ . '/../sql/migrations/081_media_pipelines.sqlite.sql') ?: '';
|
||||
if ($sql !== '') {
|
||||
$pdo->exec($sql);
|
||||
}
|
||||
});
|
||||
self::$schemaEnsured = true;
|
||||
}
|
||||
|
||||
/** @return list<array<string, mixed>> */
|
||||
public static function listForUser(int $userId, int $companyId): array {
|
||||
self::ensureSchema();
|
||||
$pdo = Database::pdo();
|
||||
$stmt = $pdo->prepare(
|
||||
'SELECT * FROM media_pipelines
|
||||
WHERE company_id = ? AND owner_user_id = ?
|
||||
ORDER BY updated_at DESC, id DESC
|
||||
LIMIT 200'
|
||||
);
|
||||
$stmt->execute([$companyId, $userId]);
|
||||
$rows = $stmt->fetchAll(PDO::FETCH_ASSOC);
|
||||
return is_array($rows) ? array_map([self::class, 'normalizeRow'], $rows) : [];
|
||||
}
|
||||
|
||||
public static function getById(int $id, int $userId, int $companyId): ?array {
|
||||
self::ensureSchema();
|
||||
$pdo = Database::pdo();
|
||||
$stmt = $pdo->prepare(
|
||||
'SELECT * FROM media_pipelines WHERE id = ? AND company_id = ? AND owner_user_id = ? LIMIT 1'
|
||||
);
|
||||
$stmt->execute([$id, $companyId, $userId]);
|
||||
$row = $stmt->fetch(PDO::FETCH_ASSOC);
|
||||
return is_array($row) ? self::normalizeRow($row) : null;
|
||||
}
|
||||
|
||||
/** @param array<string, mixed> $input */
|
||||
public static function create(int $userId, int $companyId, array $input): array {
|
||||
self::ensureSchema();
|
||||
$name = trim((string) ($input['name'] ?? 'Pipeline'));
|
||||
$sourceUrl = trim((string) ($input['source_url'] ?? ''));
|
||||
if ($sourceUrl === '') {
|
||||
throw new InvalidArgumentException('source_url required');
|
||||
}
|
||||
$pdo = Database::pdo();
|
||||
$stmt = $pdo->prepare(
|
||||
'INSERT INTO media_pipelines
|
||||
(company_id, owner_user_id, name, source_url, source_fmt, audio_source,
|
||||
filter_profile, output_preset, run_mode, status, hls_slug)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
|
||||
);
|
||||
$hlsSlug = 'p' . bin2hex(random_bytes(8));
|
||||
$stmt->execute([
|
||||
$companyId,
|
||||
$userId,
|
||||
$name !== '' ? $name : 'Pipeline',
|
||||
$sourceUrl,
|
||||
self::nullOrTrim($input['source_fmt'] ?? null),
|
||||
self::nullOrTrim($input['audio_source'] ?? null),
|
||||
self::sanitizeProfile((string) ($input['filter_profile'] ?? 'default')),
|
||||
self::sanitizePreset((string) ($input['output_preset'] ?? '1080p')),
|
||||
self::sanitizeRunMode((string) ($input['run_mode'] ?? 'full')),
|
||||
'idle',
|
||||
$hlsSlug,
|
||||
]);
|
||||
$id = (int) $pdo->lastInsertId();
|
||||
$row = self::getById($id, $userId, $companyId);
|
||||
if ($row === null) {
|
||||
throw new RuntimeException('create failed');
|
||||
}
|
||||
return $row;
|
||||
}
|
||||
|
||||
public static function updateStatus(int $id, string $status, ?string $error = null): void {
|
||||
self::ensureSchema();
|
||||
$pdo = Database::pdo();
|
||||
$stmt = $pdo->prepare(
|
||||
'UPDATE media_pipelines SET status = ?, last_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?'
|
||||
);
|
||||
$stmt->execute([$status, $error, $id]);
|
||||
}
|
||||
|
||||
/** @param array<string, mixed> $fields */
|
||||
public static function updateRuntime(int $id, array $fields): void {
|
||||
self::ensureSchema();
|
||||
$allowed = [
|
||||
'status', 'pid_encode', 'pid_fanout', 'work_dir', 'record_path',
|
||||
'hls_playlist_url', 'probe_json', 'last_error',
|
||||
];
|
||||
$sets = [];
|
||||
$vals = [];
|
||||
foreach ($allowed as $key) {
|
||||
if (array_key_exists($key, $fields)) {
|
||||
$sets[] = "$key = ?";
|
||||
$vals[] = $fields[$key];
|
||||
}
|
||||
}
|
||||
if ($sets === []) {
|
||||
return;
|
||||
}
|
||||
$sets[] = 'updated_at = CURRENT_TIMESTAMP';
|
||||
$vals[] = $id;
|
||||
$pdo = Database::pdo();
|
||||
$pdo->prepare('UPDATE media_pipelines SET ' . implode(', ', $sets) . ' WHERE id = ?')->execute($vals);
|
||||
}
|
||||
|
||||
/** @param array<string, mixed> $row */
|
||||
private static function normalizeRow(array $row): array {
|
||||
$row['id'] = (int) ($row['id'] ?? 0);
|
||||
$row['company_id'] = (int) ($row['company_id'] ?? 1);
|
||||
$row['owner_user_id'] = (int) ($row['owner_user_id'] ?? 0);
|
||||
$row['pid_encode'] = isset($row['pid_encode']) ? (int) $row['pid_encode'] : null;
|
||||
$row['pid_fanout'] = isset($row['pid_fanout']) ? (int) $row['pid_fanout'] : null;
|
||||
if (!empty($row['probe_json']) && is_string($row['probe_json'])) {
|
||||
$decoded = json_decode($row['probe_json'], true);
|
||||
$row['probe'] = is_array($decoded) ? $decoded : null;
|
||||
} else {
|
||||
$row['probe'] = null;
|
||||
}
|
||||
return $row;
|
||||
}
|
||||
|
||||
private static function nullOrTrim(mixed $v): ?string {
|
||||
if ($v === null) {
|
||||
return null;
|
||||
}
|
||||
$s = trim((string) $v);
|
||||
return $s === '' ? null : $s;
|
||||
}
|
||||
|
||||
private static function sanitizeProfile(string $p): string {
|
||||
return in_array($p, ['default', 'blurred', 'nologo'], true) ? $p : 'default';
|
||||
}
|
||||
|
||||
private static function sanitizePreset(string $p): string {
|
||||
return in_array($p, ['1080p', '720p'], true) ? $p : '1080p';
|
||||
}
|
||||
|
||||
private static function sanitizeRunMode(string $m): string {
|
||||
return $m === 'encode_only' ? 'encode_only' : 'full';
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user