<?php
/**
 * stream_api.php — DB-backed RTMP channel API + hybrid runner (PHP 5.6+ compatible)
 *
 * Fixes in this version:
 * - Derive MediaMTX input from room OR project when opts.input_url is absent.
 * - Reject base RTMP host values with no stream path, so FFmpeg never starts with:
 *     rtmp://mtx.itahukamedia.com:1935
 * - Normalize room/project into a valid per-project RTMP path.
 * - Preserve PHP 5.6 compatibility.
 *
 * Configure in .env (same folder) or server env:
 *   DB_DSN=mysql:host=...;dbname=...;charset=utf8mb4
 *   DB_USER=...
 *   DB_PASS=...
 *   STREAM_API_TOKEN=...                (optional)
 *   STREAM_INPUT_URL=...                (default full input URL; if set, must include a path)
 *   STREAM_INPUT_HOST=rtmp://mediamtx:1935  (default MediaMTX base host used to build per-room path)
 *   FFMPEG_BIN=ffmpeg                   (optional, for local spawning)
 *   CORS_ORIGINS=https://studio.itahukamedia.com,https://portal.itahukamedia.com
 *   STREAM_AGENT_URL=https://stream.itahukamedia.com   (agent base URL)
 *   STREAM_AGENT_TOKEN=...              (optional bearer to agent)
 *
 * Endpoints (path-style):
 *   POST /stream_api.php/<action>
 *   Or: POST /stream_api.php?action=<action>
 *
 * Actions (POST JSON preferred):
 *   healthz
 *   channels        {project,user_id}
 *   save_channel    {project,user_id,channel:{id,name,rtmp_url,rtmp_key,enabled?}}
 *   rename_channel  {project,user_id,channel:{id},name}
 *   set_enabled     {project,user_id,channel:{id},enabled}
 *   remove_channel  {project,user_id,channel:{id}}
 *   reveal_channel  {project,user_id,channel:{id}}
 *   test            {channel:{rtmp_url,rtmp_key}}
 *   start           {project,user_id,room?,channel:{id}|channel_id?,opts:{input_url?,audio_only?,extra_args?,force_agent?}}
 *   stop            {project,user_id,room?,channel:{id}|channel_id?}
 *   stop_project    {project,user_id}
 *   status          {project,user_id,channels:[{id} or "id"]}
 */

error_reporting(E_ALL);
ini_set('display_errors', '0');

// Prevent browser/proxy caching (helps avoid serving an older stream_api.php)
header('Cache-Control: no-store, no-cache, must-revalidate, max-age=0');
header('Pragma: no-cache');
header('Expires: 0');

set_error_handler(function($sev,$msg,$file,$line){
  if (!(error_reporting() & $sev)) return false;
  error_log("[stream_api] $msg in $file:$line");
  return true;
});

register_shutdown_function(function(){
  $e = error_get_last();
  if (!$e) return;
  $fatal = in_array($e['type'], array(E_ERROR,E_PARSE,E_CORE_ERROR,E_COMPILE_ERROR), true);
  if (!$fatal) return;
  if (headers_sent()) return;
  http_response_code(500);
  header('Content-Type: application/json; charset=utf-8');
  echo json_encode(array('ok'=>false,'error'=>'fatal','detail'=>$e['message']));
});

if (!defined('STATE_DIR')) {
  $candidates = array(
    __DIR__ . '/.stream_state',
    rtrim(sys_get_temp_dir(), DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR . 'itahuka_stream_state',
  );
  $chosen = null;
  foreach ($candidates as $dir) {
    @mkdir($dir, 0775, true);
    if (is_dir($dir) && is_writable($dir)) { $chosen = $dir; break; }
  }
  if ($chosen === null) $chosen = __DIR__;
  define('STATE_DIR', $chosen);
}

header('Content-Type: application/json; charset=utf-8');

/* -----------------------------
   Session-bound user id
----------------------------- */
$USER_SESSION_KEYS = array('user_id','uid','auth_user_id','account_id','user','userid');

function ensure_session_started() {
  if (PHP_SAPI === 'cli') return;
  if (function_exists('session_status')) {
    if (session_status() === PHP_SESSION_ACTIVE) return;
  } else {
    if (session_id() !== '') return;
  }
  if (headers_sent()) return;
  @session_start();
}
function session_user_id($USER_SESSION_KEYS) {
  ensure_session_started();
  if (!isset($_SESSION) || !is_array($_SESSION)) return null;
  foreach ($USER_SESSION_KEYS as $k) {
    if (isset($_SESSION[$k]) && is_scalar($_SESSION[$k])) {
      $v = trim((string)$_SESSION[$k]);
      if ($v !== '') return $v;
    }
  }
  return null;
}

/* -----------------------------
   .env loader (minimal)
----------------------------- */
function load_local_env($path) {
  if (!is_file($path) || !is_readable($path)) return;
  $fh = @fopen($path, 'rb'); if (!$fh) return;
  while (!feof($fh)) {
    $line = fgets($fh);
    if ($line === false) break;
    $line = trim($line);
    if ($line === '' || $line[0] === '#') continue;
    if (stripos($line, 'export ') === 0) $line = trim(substr($line, 7));
    $pos = strpos($line, '=');
    if ($pos === false) continue;
    $key = trim(substr($line, 0, $pos));
    $val = trim(substr($line, $pos + 1));
    if ($key === '') continue;

    $len = strlen($val);
    if ($len >= 2) {
      $a = $val[0]; $b = $val[$len-1];
      if (($a === '"' && $b === '"') || ($a === "'" && $b === "'")) $val = substr($val, 1, -1);
    }

    $existing = getenv($key);
    if ($existing !== false && $existing !== '') continue;
    @putenv($key . '=' . $val);
    $_ENV[$key] = $val;
  }
  @fclose($fh);
}
$envPath = __DIR__ . '/.env';
$envSz = @filesize($envPath);
if (is_int($envSz) && $envSz <= 1024*1024) load_local_env($envPath);

/* -----------------------------
   CORS
----------------------------- */
$allowedOrigins = array();
$rawOrigins = getenv('CORS_ORIGINS');
if ($rawOrigins !== false && trim($rawOrigins) !== '') {
  foreach (explode(',', $rawOrigins) as $o) {
    $o = trim($o);
    if ($o !== '') $allowedOrigins[] = $o;
  }
}
$origin = isset($_SERVER['HTTP_ORIGIN']) ? (string)$_SERVER['HTTP_ORIGIN'] : '';
if ($origin !== '' && (empty($allowedOrigins) || in_array($origin, $allowedOrigins, true))) {
  header("Access-Control-Allow-Origin: {$origin}");
  header('Vary: Origin');
}
header('Access-Control-Allow-Methods: GET, POST, OPTIONS');
header('Access-Control-Allow-Headers: Content-Type, Authorization, X-Stream-Token');
header('Access-Control-Max-Age: 600');

if (isset($_SERVER['REQUEST_METHOD']) && $_SERVER['REQUEST_METHOD'] === 'OPTIONS') {
  http_response_code(204);
  exit;
}

/* -----------------------------
   Helpers
----------------------------- */
function json_out($code, $data) {
  http_response_code((int)$code);
  echo json_encode($data, JSON_UNESCAPED_SLASHES);
  exit;
}
function read_json_body() {
  $raw = file_get_contents('php://input');
  if (!is_string($raw) || trim($raw) === '') return array();
  $j = json_decode($raw, true);
  return is_array($j) ? $j : array();
}
function safe_id($id) {
  $id = preg_replace('~[^a-zA-Z0-9_\-\.]~', '_', trim((string)$id));
  $id = trim($id, '_');
  return $id !== '' ? $id : '';
}
function safe_project($p) {
  $p = trim((string)$p);
  if ($p === '') $p = 'default';
  $p = preg_replace('~[^a-zA-Z0-9_\-]~', '_', $p);
  $p = trim($p, '_');
  return $p !== '' ? $p : 'default';
}
function endpoint_action() {
  if (isset($_GET['action']) && is_string($_GET['action']) && trim($_GET['action']) !== '') {
    return safe_id($_GET['action']);
  }
  $path = isset($_SERVER['PATH_INFO']) ? trim((string)$_SERVER['PATH_INFO'], '/') : '';
  if ($path !== '') return safe_id($path);
  $uri = isset($_SERVER['REQUEST_URI']) ? (string)$_SERVER['REQUEST_URI'] : '';
  if (preg_match('~/(stream_api(?:_[^/]+)?\.php)/([a-zA-Z0-9_\-]+)~', $uri, $m)) {
    return safe_id($m[2]);
  }
  return '';
}
function read_bearer_token($payload) {
  $auth = isset($_SERVER['HTTP_AUTHORIZATION']) ? (string)$_SERVER['HTTP_AUTHORIZATION'] : '';
  if ($auth && preg_match('~^\s*Bearer\s+(.+)\s*$~i', $auth, $m)) return trim($m[1]);
  $xt = isset($_SERVER['HTTP_X_STREAM_TOKEN']) ? (string)$_SERVER['HTTP_X_STREAM_TOKEN'] : '';
  if (trim($xt) !== '') return trim($xt);
  foreach (array('token','api_token','stream_token') as $k) {
    if (isset($payload[$k]) && is_string($payload[$k]) && trim($payload[$k]) !== '') return trim($payload[$k]);
  }
  return '';
}
function require_token_or_401($requiredToken, $givenToken) {
  $requiredToken = trim((string)$requiredToken);
  if ($requiredToken === '') return;
  if ($givenToken === '' || !hash_equals($requiredToken, $givenToken)) json_out(401, array('ok'=>false,'error'=>'unauthorized'));
}
function normalize_rtmp_pair($url, $key) {
  $url = trim((string)$url);
  $key = trim((string)$key);
  // If key embedded already, accept as-is.
  if ($key === '' && preg_match('~^rtmps?://~i', $url)) return array($url, '');
  return array($url, $key);
}
function build_rtmp_target($rtmpUrl, $rtmpKey) {
  $rtmpUrl = trim((string)$rtmpUrl);
  $rtmpKey = trim((string)$rtmpKey);
  if (!preg_match('~^rtmps?://~i', $rtmpUrl)) throw new Exception("rtmp_url must start with rtmp:// or rtmps://");
  if ($rtmpKey !== '') return rtrim($rtmpUrl, '/') . '/' . ltrim($rtmpKey, '/');
  return $rtmpUrl;
}
function is_base_rtmp_host_only($u) {
  $u = trim((string)$u);
  if ($u === '') return true;
  $parts = @parse_url($u);
  if (!is_array($parts)) return false;
  $scheme = isset($parts['scheme']) ? strtolower((string)$parts['scheme']) : '';
  if ($scheme !== 'rtmp' && $scheme !== 'rtmps') return false;
  $path = isset($parts['path']) ? trim((string)$parts['path']) : '';
  return ($path === '' || $path === '/');
}
function normalize_room_name($roomOrProject) {
  $v = trim((string)$roomOrProject);
  if ($v === '') return '';
  $v = preg_replace('~[^A-Za-z0-9_\-\.]~', '_', $v);
  $v = trim($v, '_');
  return $v;
}
function derive_room_name($payload) {
  $room = '';
  if (isset($payload['room'])) $room = normalize_room_name($payload['room']);
  if ($room !== '') return $room;

  if (isset($payload['project'])) $room = normalize_room_name($payload['project']);
  if ($room !== '') return $room;

  if (isset($payload['project_id'])) $room = normalize_room_name($payload['project_id']);
  if ($room !== '') return $room;

  return '';
}

/* -----------------------------
   DB
----------------------------- */
function db() {
  static $pdo = null; static $tried = false;
  if ($tried) return $pdo;
  $tried = true;
  $dsn = getenv('DB_DSN'); $dsn = ($dsn === false) ? '' : (string)$dsn;
  if ($dsn === '') return null;
  $user = getenv('DB_USER'); $user = ($user === false) ? '' : (string)$user;
  $pass = getenv('DB_PASS'); $pass = ($pass === false) ? '' : (string)$pass;
  try {
    $pdo = new PDO($dsn, $user, $pass, array(
      PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
      PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
    ));
    return $pdo;
  } catch (Exception $e) {
    error_log("[stream_api] DB connect failed: " . $e->getMessage());
    return null;
  }
}
function has_column($pdo, $table, $col) {
  static $cache = array();
  $k = $table . '.' . $col;
  if (isset($cache[$k])) return $cache[$k];
  try {
    $st = $pdo->prepare("SHOW COLUMNS FROM `$table` LIKE :c");
    $st->execute(array(':c'=>$col));
    $ok = (bool)$st->fetch();
    $cache[$k] = $ok;
    return $ok;
  } catch (Exception $e) {
    $cache[$k] = false;
    return false;
  }
}
function db_exec($pdo, $sql, $params) {
  $st = $pdo->prepare($sql);
  $st->execute($params);
}

/* -----------------------------
   State (local PID tracking)
----------------------------- */
function state_key($project, $userId, $channelId) {
  $uid = ($userId !== null && trim((string)$userId) !== '') ? safe_id($userId) : 'anon';
  $cid = safe_id($channelId);
  return $project . '__' . $uid . '__' . $cid;
}
function state_path($stateKey) {
  return rtrim(STATE_DIR, DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR . $stateKey . '.json';
}
function load_state($stateKey) {
  $p = state_path($stateKey);
  if (!is_file($p)) return array();
  $raw = @file_get_contents($p);
  if (!is_string($raw)) return array();
  $j = json_decode($raw, true);
  return is_array($j) ? $j : array();
}
function save_state($stateKey, $state) {
  $p = state_path($stateKey);
  @file_put_contents($p, json_encode($state, JSON_UNESCAPED_SLASHES));
}
function is_proc_running($pid) {
  $pid = (int)$pid;
  if ($pid <= 1) return false;
  if (stripos(PHP_OS, 'WIN') === 0) return false;
  if (function_exists('posix_kill')) return @posix_kill($pid, 0);
  @exec('kill -0 ' . $pid . ' 2>/dev/null', $o, $rc);
  return $rc === 0;
}
function can_local_spawn() {
  if (stripos(PHP_OS, 'WIN') === 0) return false;
  if (!function_exists('shell_exec')) return false;
  $dis = ini_get('disable_functions');
  if (is_string($dis) && $dis !== '') {
    $arr = array_map('trim', explode(',', $dis));
    if (in_array('shell_exec', $arr, true) || in_array('exec', $arr, true)) return false;
  }
  $ff = getenv('FFMPEG_BIN'); $ff = ($ff === false || trim($ff) === '') ? 'ffmpeg' : trim($ff);
  $which = @shell_exec("command -v " . escapeshellarg($ff) . " 2>/dev/null");
  return is_string($which) && trim($which) !== '';
}
function local_spawn_ffmpeg($cid, $input, $target, $opts) {
  $ff = getenv('FFMPEG_BIN'); $ff = ($ff === false || trim($ff) === '') ? 'ffmpeg' : trim($ff);
  $audioOnly = !empty($opts['audio_only']);
  $extra = '';
  if (isset($opts['ffmpeg_args'])) $extra = (string)$opts['ffmpeg_args'];
  if ($extra === '' && isset($opts['extra_args'])) $extra = (string)$opts['extra_args'];
  $extra = trim($extra);

  $log = rtrim(STATE_DIR, DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR . "ffmpeg_{$cid}_" . date('Ymd_His') . ".log";
  $cmd = escapeshellcmd($ff)
    . " -hide_banner -nostdin -re -i " . escapeshellarg($input)
    . ($audioOnly ? " -vn -c:a aac -b:a 128k -ar 48000 -ac 2" :
        " -c:v libx264 -preset veryfast -tune zerolatency -profile:v main -level 4.1 -pix_fmt yuv420p -b:v 3000k -maxrate 3500k -bufsize 6000k -g 60 -keyint_min 60 -c:a aac -b:a 160k -ar 48000 -ac 2")
    . ($extra !== '' ? " " . $extra : "")
    . " -f flv " . escapeshellarg($target)
    . " >> " . escapeshellarg($log) . " 2>&1 & echo $!";
  $pidRaw = @shell_exec($cmd);
  $pid = (int)trim((string)$pidRaw);
  if ($pid <= 1) throw new Exception('Failed to spawn FFmpeg (pid invalid).');
  return array('pid'=>$pid, 'cmd'=>$cmd, 'log'=>$log);
}
function local_stop_ffmpeg($pid) {
  $pid = (int)$pid;
  if ($pid <= 1) return;
  if (function_exists('posix_kill')) {
    @posix_kill($pid, 15); usleep(400000);
    if (@posix_kill($pid, 0)) @posix_kill($pid, 9);
    return;
  }
  @exec('kill ' . $pid . ' 2>/dev/null');
}

/* -----------------------------
   Agent runner (docker stream-api)
----------------------------- */
function agent_base_url() {
  $u = getenv('STREAM_AGENT_URL');
  $u = ($u === false) ? '' : trim((string)$u);
  return $u === '' ? '' : rtrim($u, '/');
}
function agent_call($path, $payload) {
  $base = agent_base_url();
  if ($base === '') throw new Exception('Agent runner not configured (STREAM_AGENT_URL missing).');
  $url = $base . '/' . ltrim($path, '/');
  $token = getenv('STREAM_AGENT_TOKEN');
  $token = ($token === false) ? '' : trim((string)$token);

  $headers = array('Content-Type: application/json', 'Accept: application/json');
  if ($token !== '') $headers[] = 'Authorization: Bearer ' . $token;

  $ctx = stream_context_create(array('http'=>array(
    'method'=>'POST',
    'header'=>implode("\r\n",$headers),
    'content'=>json_encode($payload, JSON_UNESCAPED_SLASHES),
    'timeout'=>20,
  )));
  $raw = @file_get_contents($url, false, $ctx);
  if ($raw === false) {
    $err = error_get_last();
    throw new Exception('Agent call failed: ' . (isset($err['message']) ? $err['message'] : 'unknown error'));
  }
  $j = json_decode($raw, true);
  if (!is_array($j)) throw new Exception('Agent returned non-JSON response.');
  return $j;
}
function resolve_channel_id($payload) {
  if (isset($payload['channel']) && is_string($payload['channel'])) return safe_id($payload['channel']);
  $ch = (isset($payload['channel']) && is_array($payload['channel'])) ? $payload['channel'] : array();
  $id = '';
  if (isset($ch['id'])) $id = (string)$ch['id'];
  if ($id === '' && isset($payload['channel_id'])) $id = (string)$payload['channel_id'];
  if ($id === '' && isset($payload['id'])) $id = (string)$payload['id'];
  if ($id === '' && isset($ch['name'])) $id = (string)$ch['name'];
  if ($id === '' && isset($payload['name'])) $id = (string)$payload['name'];
  return safe_id($id);
}
function default_input_url($payload) {
  $opts = (isset($payload['opts']) && is_array($payload['opts'])) ? $payload['opts'] : array();
  $u = '';
  if (isset($opts['input_url'])) $u = (string)$opts['input_url'];
  if ($u === '' && isset($payload['input'])) $u = (string)$payload['input'];
  if ($u === '' && isset($payload['input_url'])) $u = (string)$payload['input_url'];

  $u = trim($u);

  // Reject base RTMP host values with no stream path.
  if ($u !== '' && is_base_rtmp_host_only($u)) {
    throw new Exception('input_url is missing RTMP stream path. Expected something like rtmp://mediamtx:1935/room_proj_xxx');
  }

  // If input_url is absent, build it from room OR project.
  if ($u === '') {
    $room = derive_room_name($payload);
    if ($room !== '') {
      $h = getenv('STREAM_INPUT_HOST');
      if ($h === false || trim($h) === '') $h = 'rtmp://mediamtx:1935';
      $h = rtrim(trim($h), '/');
      $u = $h . '/' . $room;
    }
  }

  // Only use STREAM_INPUT_URL if it is a FULL URL with a path.
  if ($u === '') {
    $e = getenv('STREAM_INPUT_URL');
    $u = ($e === false) ? '' : trim((string)$e);
    if ($u !== '' && is_base_rtmp_host_only($u)) {
      throw new Exception('STREAM_INPUT_URL is set but missing RTMP stream path. Use STREAM_INPUT_HOST for base host, or set STREAM_INPUT_URL to a full path.');
    }
  }

  $u = trim($u);
  if ($u === '') throw new Exception('Missing input_url. Provide opts.input_url, or provide room/project so the API can derive a per-room MediaMTX path.');

  return $u;
}

/* -----------------------------
   Dispatch
----------------------------- */
$action = endpoint_action();
$payload = array();
if (is_array($_GET)) $payload = array_merge($payload, $_GET);
if (is_array($_POST)) $payload = array_merge($payload, $_POST);
$body = read_json_body();
if (is_array($body)) $payload = array_merge($payload, $body);

// --- STREAM UI action aliases (coproducer.bundle.js) ---
// The Stream UI uses: add_channel, update_channel, delete_channel, stop_all.
// Map them to canonical actions implemented below and normalize payload shape.
if ($action === 'add_channel') {
  // UI sends: { name, platform, rtmp_url, stream_key }
  $name = isset($payload['name']) ? trim((string)$payload['name']) : '';
  if ($name === '') json_out(400, array('ok'=>false,'error'=>'missing name'));
  $cid = safe_id($name);
  if ($cid === '') $cid = 'ch_' . substr(md5($name . microtime(true)), 0, 10);

  $platform = isset($payload['platform']) ? trim((string)$payload['platform']) : '';
  if ($platform === '') $platform = $name;

  $rtmpUrl = isset($payload['rtmp_url']) ? trim((string)$payload['rtmp_url']) : '';
  if ($rtmpUrl === '' && isset($payload['stream_url'])) $rtmpUrl = trim((string)$payload['stream_url']);

  $rtmpKey = '';
  if (isset($payload['stream_key'])) $rtmpKey = trim((string)$payload['stream_key']);
  if ($rtmpKey === '' && isset($payload['rtmp_key'])) $rtmpKey = trim((string)$payload['rtmp_key']);

  $payload['channel'] = array(
    'id' => $cid,
    'name' => $name,
    'channel_name' => $name,
    'platform' => $platform,
    'rtmp_url' => $rtmpUrl,
    'rtmp_key' => $rtmpKey,
    'enabled' => 1,
  );
  $payload['channel_id'] = $cid;
  $action = 'save_channel';
}

if ($action === 'update_channel') {
  // UI sends: { channel_id, name? , platform? , rtmp_url? , stream_key? , enabled? }
  $cid = isset($payload['channel_id']) ? safe_id((string)$payload['channel_id']) : '';
  if ($cid === '') $cid = resolve_channel_id($payload);
  if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel_id'));

  $ch = array('id'=>$cid);

  if (isset($payload['name'])) $ch['name'] = trim((string)$payload['name']);
  if (isset($payload['channel_name'])) $ch['name'] = trim((string)$payload['channel_name']);
  if (isset($payload['platform'])) $ch['platform'] = trim((string)$payload['platform']);
  if (isset($payload['rtmp_url'])) $ch['rtmp_url'] = trim((string)$payload['rtmp_url']);
  if (isset($payload['stream_url'])) $ch['rtmp_url'] = trim((string)$payload['stream_url']);

  $k = '';
  if (isset($payload['stream_key'])) $k = trim((string)$payload['stream_key']);
  if ($k === '' && isset($payload['rtmp_key'])) $k = trim((string)$payload['rtmp_key']);
  if ($k !== '') $ch['rtmp_key'] = $k;

  if (isset($payload['enabled'])) $ch['enabled'] = (int)!!$payload['enabled'];

  $payload['channel'] = $ch;
  $payload['channel_id'] = $cid;
  $action = 'save_channel';
}

if ($action === 'delete_channel') {
  $cid = isset($payload['channel_id']) ? safe_id((string)$payload['channel_id']) : '';
  if ($cid === '') $cid = resolve_channel_id($payload);
  if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel_id'));
  $payload['channel'] = array('id'=>$cid);
  $payload['channel_id'] = $cid;
  $action = 'remove_channel';
}

if ($action === 'stop_all') {
  $action = 'stop_project';
}
// --- end aliases ---


$requiredToken = getenv('STREAM_API_TOKEN');
$requiredToken = ($requiredToken === false) ? '' : (string)$requiredToken;
$givenToken = read_bearer_token($payload);
require_token_or_401($requiredToken, $givenToken);

if ($action === '') json_out(400, array('ok'=>false,'error'=>'missing action'));

$pdo = db();

try {
  if ($action === 'healthz') json_out(200, array('ok'=>true));

  if ($action === 'test') {
    $ch = (isset($payload['channel']) && is_array($payload['channel'])) ? $payload['channel'] : array();
    $u = '';
    if (isset($ch['rtmp_url'])) $u = (string)$ch['rtmp_url'];
    if ($u === '' && isset($payload['rtmp_url'])) $u = (string)$payload['rtmp_url'];
    if ($u === '' && isset($ch['url'])) $u = (string)$ch['url'];
    $k = '';
    if (isset($ch['rtmp_key'])) $k = (string)$ch['rtmp_key'];
    if ($k === '' && isset($payload['rtmp_key'])) $k = (string)$payload['rtmp_key'];
    if ($k === '' && isset($ch['key'])) $k = (string)$ch['key'];
    $target = build_rtmp_target($u, $k);
    json_out(200, array('ok'=>true,'target'=>$target));
  }

  $project = safe_project(isset($payload['project']) ? $payload['project'] : (isset($payload['project_id']) ? $payload['project_id'] : (isset($payload['room']) ? $payload['room'] : 'default')));
  $room = derive_room_name($payload);

  global $USER_SESSION_KEYS;
  $sessionUid = session_user_id($USER_SESSION_KEYS);
  $userId = ($sessionUid !== null) ? $sessionUid : (isset($payload['user_id']) ? (string)$payload['user_id'] : null);

  // DB-required actions enforce userId
  $needsUser = in_array($action, array('start','stop','stop_project','save_channel','remove_channel','reveal_channel','rename_channel','set_enabled','channels','status'), true);
  if ($pdo && $needsUser) {
    if ($userId === null || trim($userId) === '') {
      json_out(400, array('ok'=>false,'error'=>'missing user_id','hint'=>'Login to establish a session, or pass user_id (server-to-server only).'));
    }
  }

  if ($action === 'channels') {
    $channels = array();
    if ($pdo) {
      $uid = $userId;
      $nameCol = has_column($pdo, 'stream_channels', 'name') ? 'name' : (has_column($pdo, 'stream_channels', 'channel_name') ? 'channel_name' : 'channel_id');
      $platformCol = has_column($pdo, 'stream_channels', 'platform') ? 'platform' : "''";
      $enabledCol = has_column($pdo, 'stream_channels', 'enabled') ? 'enabled' : '1';
      $sortCol = has_column($pdo, 'stream_channels', 'sort_order') ? 'sort_order' : (has_column($pdo, 'stream_channels', 'id') ? 'id' : 'channel_id');

      $sql = "SELECT channel_id, {$nameCol} AS name, {$platformCol} AS platform, rtmp_url, {$enabledCol} AS enabled
              FROM stream_channels
              WHERE user_id=:uid AND project=:p
              ORDER BY {$sortCol} ASC
              LIMIT 500";
      $st = $pdo->prepare($sql);
      $st->execute(array(':uid'=>$uid, ':p'=>$project));
      foreach ($st->fetchAll() as $r) {
        $channels[] = array(
          'id'=>(string)(isset($r['channel_id']) ? $r['channel_id'] : ''),
          'name'=>(string)(isset($r['name']) ? $r['name'] : (isset($r['channel_id']) ? $r['channel_id'] : '')),
          'platform'=>(string)(isset($r['platform']) ? $r['platform'] : ''),
          'rtmp_url'=>(string)(isset($r['rtmp_url']) ? $r['rtmp_url'] : ''),
          'enabled'=>(int)(isset($r['enabled']) ? $r['enabled'] : 1),
          'rtmp_key_masked'=>'********',
        );
      }
    }
    json_out(200, array('ok'=>true,'project'=>$project,'channels'=>$channels,'db_enabled'=>($pdo?true:false)));
  }

  if ($action === 'rename_channel') {
    if (!$pdo) json_out(500, array('ok'=>false,'error'=>'db not configured'));
    $cid = resolve_channel_id($payload);
    if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel.id'));
    $newName = '';
    if (isset($payload['name'])) $newName = trim((string)$payload['name']);
    if ($newName === '' && isset($payload['channel']) && is_array($payload['channel']) && isset($payload['channel']['name'])) $newName = trim((string)$payload['channel']['name']);
    if ($newName === '') json_out(400, array('ok'=>false,'error'=>'missing name'));

    if (has_column($pdo, 'stream_channels', 'name')) {
      db_exec($pdo, "UPDATE stream_channels SET name=:n WHERE user_id=:uid AND project=:p AND channel_id=:cid",
        array(':n'=>$newName, ':uid'=>$userId, ':p'=>$project, ':cid'=>$cid));
    } else {
      db_exec($pdo, "UPDATE stream_channels SET channel_name=:n WHERE user_id=:uid AND project=:p AND channel_id=:cid",
        array(':n'=>$newName, ':uid'=>$userId, ':p'=>$project, ':cid'=>$cid));
    }
    json_out(200, array('ok'=>true,'id'=>$cid,'name'=>$newName));
  }

  if ($action === 'set_enabled') {
    if (!$pdo) json_out(500, array('ok'=>false,'error'=>'db not configured'));
    $cid = resolve_channel_id($payload);
    if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel.id'));
    $enabled = 1;
    if (isset($payload['enabled'])) $enabled = (int)!!$payload['enabled'];
    if (isset($payload['enable'])) $enabled = (int)!!$payload['enable'];
    if (!has_column($pdo, 'stream_channels', 'enabled')) json_out(400, array('ok'=>false,'error'=>'stream_channels.enabled column missing'));
    db_exec($pdo, "UPDATE stream_channels SET enabled=:e WHERE user_id=:uid AND project=:p AND channel_id=:cid",
      array(':e'=>$enabled, ':uid'=>$userId, ':p'=>$project, ':cid'=>$cid));
    json_out(200, array('ok'=>true,'id'=>$cid,'enabled'=>$enabled));
  }

  if ($action === 'save_channel') {
    if (!$pdo) json_out(500, array('ok'=>false,'error'=>'db not configured'));
    $ch = (isset($payload['channel']) && is_array($payload['channel'])) ? $payload['channel'] : array();
    $id = '';
    if (isset($ch['id'])) $id = (string)$ch['id'];
    if ($id === '' && isset($payload['channel_id'])) $id = (string)$payload['channel_id'];
    if ($id === '' && isset($ch['name'])) $id = (string)$ch['name'];
    if ($id === '' && isset($payload['name'])) $id = (string)$payload['name'];
    $id = safe_id($id);
    if ($id === '') json_out(400, array('ok'=>false,'error'=>'missing channel.id'));

    $name = '';
    if (isset($ch['name'])) $name = (string)$ch['name'];
    if ($name === '' && isset($payload['channel_name'])) $name = (string)$payload['channel_name'];
    if ($name === '') $name = $id;
    $name = trim($name);

    $platform = '';
    if (isset($ch['platform'])) $platform = (string)$ch['platform'];
    if ($platform === '' && isset($payload['platform'])) $platform = (string)$payload['platform'];
    $platform = trim($platform);

    $url = '';
    if (isset($ch['rtmp_url'])) $url = (string)$ch['rtmp_url'];
    if ($url === '' && isset($ch['stream_url'])) $url = (string)$ch['stream_url'];
    if ($url === '' && isset($payload['rtmp_url'])) $url = (string)$payload['rtmp_url'];
    if ($url === '' && isset($payload['stream_url'])) $url = (string)$payload['stream_url'];

    $key = '';
    if (isset($ch['rtmp_key'])) $key = (string)$ch['rtmp_key'];
    if ($key === '' && isset($ch['stream_key'])) $key = (string)$ch['stream_key'];
    if ($key === '' && isset($payload['rtmp_key'])) $key = (string)$payload['rtmp_key'];
    if ($key === '' && isset($payload['stream_key'])) $key = (string)$payload['stream_key'];

    $pair = normalize_rtmp_pair($url, $key);
    $rtmpUrl = $pair[0]; $rtmpKey = $pair[1];
    if ($rtmpUrl === '') json_out(400, array('ok'=>false,'error'=>'missing rtmp_url'));
    build_rtmp_target($rtmpUrl, $rtmpKey);

    $enabled = 1;
    if (isset($ch['enabled'])) $enabled = (int)!!$ch['enabled'];
    if (isset($payload['enabled'])) $enabled = (int)!!$payload['enabled'];

    $nameCol = has_column($pdo, 'stream_channels', 'name') ? 'name' : 'channel_name';
    $enabledCol = has_column($pdo, 'stream_channels', 'enabled') ? 'enabled' : null;
    $platformCol = has_column($pdo, 'stream_channels', 'platform') ? 'platform' : null;

    $exists = $pdo->prepare("SELECT channel_id FROM stream_channels WHERE user_id=:uid AND project=:p AND channel_id=:cid LIMIT 1");
    $exists->execute(array(':uid'=>$userId, ':p'=>$project, ':cid'=>$id));
    $has = (bool)$exists->fetch();

    if ($has) {
      $sql = "UPDATE stream_channels SET {$nameCol}=:n, rtmp_url=:u, rtmp_key=:k" . ($platformCol ? ", {$platformCol}=:pl" : "") . "" . ($enabledCol ? ", {$enabledCol}=:e" : "") . " WHERE user_id=:uid AND project=:p AND channel_id=:cid";
      $params = array(':n'=>$name, ':u'=>$rtmpUrl, ':k'=>$rtmpKey, ':uid'=>$userId, ':p'=>$project, ':cid'=>$id);
      if ($platformCol) $params[':pl'] = $platform;
      if ($enabledCol) $params[':e'] = $enabled;
      db_exec($pdo, $sql, $params);
    } else {
      $cols = "user_id, project, channel_id, {$nameCol}, rtmp_url, rtmp_key";
      $vals = ":uid, :p, :cid, :n, :u, :k";
      $params = array(':uid'=>$userId, ':p'=>$project, ':cid'=>$id, ':n'=>$name, ':u'=>$rtmpUrl, ':k'=>$rtmpKey);
      if ($platformCol) { $cols .= ", {$platformCol}"; $vals .= ", :pl"; $params[':pl'] = $platform; }
      if ($enabledCol) { $cols .= ", {$enabledCol}"; $vals .= ", :e"; $params[':e'] = $enabled; }
      db_exec($pdo, "INSERT INTO stream_channels ({$cols}) VALUES ({$vals})", $params);
    }

    json_out(200, array('ok'=>true,'id'=>$id,'name'=>$name,'rtmp_url'=>$rtmpUrl,'enabled'=>$enabled));
  }

  if ($action === 'remove_channel') {
    if (!$pdo) json_out(500, array('ok'=>false,'error'=>'db not configured'));
    $cid = resolve_channel_id($payload);
    if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel.id'));
    db_exec($pdo, "DELETE FROM stream_channels WHERE user_id=:uid AND project=:p AND channel_id=:cid",
      array(':uid'=>$userId, ':p'=>$project, ':cid'=>$cid));
    json_out(200, array('ok'=>true,'id'=>$cid));
  }

  if ($action === 'reveal_channel') {
    if (!$pdo) json_out(500, array('ok'=>false,'error'=>'db not configured'));
    $cid = resolve_channel_id($payload);
    if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel.id'));
    $st = $pdo->prepare("SELECT channel_id, channel_name, rtmp_url, rtmp_key FROM stream_channels WHERE user_id=:uid AND project=:p AND channel_id=:cid LIMIT 1");
    $st->execute(array(':uid'=>$userId, ':p'=>$project, ':cid'=>$cid));
    $r = $st->fetch();
    if (!$r) json_out(404, array('ok'=>false,'error'=>'not found'));
    json_out(200, array('ok'=>true,'channel'=>array(
      'id'=>(string)$r['channel_id'],
      'name'=>(string)(isset($r['channel_name']) ? $r['channel_name'] : $r['channel_id']),
      'rtmp_url'=>(string)(isset($r['rtmp_url']) ? $r['rtmp_url'] : ''),
      'rtmp_key'=>(string)(isset($r['rtmp_key']) ? $r['rtmp_key'] : ''),
    )));
  }

  if ($action === 'start') {
    if (!$pdo) json_out(500, array('ok'=>false,'error'=>'db not configured (DB_DSN missing or connection failed)'));
    $cid = resolve_channel_id($payload);
    if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel.id'));
    $stateKey = state_key($project, $userId, $cid);

    $st = $pdo->prepare("SELECT rtmp_url, rtmp_key FROM stream_channels WHERE user_id=:uid AND project=:p AND channel_id=:cid LIMIT 1");
    $st->execute(array(':uid'=>$userId, ':p'=>$project, ':cid'=>$cid));
    $row = $st->fetch();
    if (!$row) json_out(404, array('ok'=>false,'error'=>'channel not found (save channel first)','id'=>$cid));

    $target = build_rtmp_target(isset($row['rtmp_url']) ? $row['rtmp_url'] : '', isset($row['rtmp_key']) ? $row['rtmp_key'] : '');
    $input = default_input_url($payload);
    $opts = (isset($payload['opts']) && is_array($payload['opts'])) ? $payload['opts'] : array();

    $stLocal = load_state($stateKey);
    if (isset($stLocal['pid']) && is_proc_running((int)$stLocal['pid'])) {
      json_out(200, array('ok'=>true,'id'=>$cid,'state'=>'RUNNING','pid'=>(int)$stLocal['pid'],'mode'=>(isset($stLocal['mode'])?$stLocal['mode']:'local'),'message'=>'already running'));
    }

    $forceAgent = (!empty($opts['force_agent']) || (isset($opts['runner']) && (string)$opts['runner'] === 'agent'));

    if (!$forceAgent && can_local_spawn()) {
      $proc = local_spawn_ffmpeg($cid, $input, $target, $opts);
      $state = array(
        'id'=>$cid,'state_key'=>$stateKey,'project'=>$project,'user_id'=>$userId,'room'=>$room,
        'state'=>'RUNNING','mode'=>'local','pid'=>(int)$proc['pid'],
        'target'=>$target,'input'=>$input,'cmd'=>$proc['cmd'],'log'=>$proc['log'],
        'started_at'=>date('c'),
      );
      save_state($stateKey, $state);
      json_out(200, array('ok'=>true,'id'=>$cid,'state'=>'RUNNING','mode'=>'local','pid'=>$state['pid'],'log'=>$state['log'],'input'=>$input));
    }

    $extraArgs = '';
    if (isset($opts['extra_args'])) $extraArgs = (string)$opts['extra_args'];
    if ($extraArgs === '' && isset($opts['ffmpeg_args'])) $extraArgs = (string)$opts['ffmpeg_args'];

    $resolvedRoom = ($room !== '' ? $room : $project);

    $resp = agent_call('/start', array(
      'room' => $resolvedRoom,
      'channel' => array(
        'id' => $cid,
        'name' => $cid,
        'enabled' => true,
        'rtmp_url' => (string)$row['rtmp_url'],
        'key' => (string)$row['rtmp_key'],
      ),
      'opts' => array(
        'input_url' => $input,
        'audio_only' => (!empty($opts['audio_only'])),
        'extra_args' => $extraArgs,
      ),
    ));
    if (!isset($resp['ok']) || !$resp['ok']) throw new Exception(isset($resp['error']) ? (string)$resp['error'] : 'agent start failed');

    $state = array(
      'id'=>$cid,'state_key'=>$stateKey,'project'=>$project,'user_id'=>$userId,'room'=>$resolvedRoom,
      'state'=>'RUNNING','mode'=>'agent','pid'=>(int)(isset($resp['pid']) ? $resp['pid'] : 0),
      'log'=>(string)(isset($resp['log']) ? $resp['log'] : ''),
      'target'=>$target,'input'=>$input,'started_at'=>date('c'),
    );
    save_state($stateKey, $state);

    $resp['mode'] = 'agent';
    $resp['id'] = $cid;
    $resp['input'] = $input;
    json_out(200, $resp);
  }

  if ($action === 'stop') {
    $cid = resolve_channel_id($payload);
    if ($cid === '') json_out(400, array('ok'=>false,'error'=>'missing channel.id'));

    $stateKey = state_key($project, $userId, $cid);
    $stLocal = load_state($stateKey);
    $mode = isset($stLocal['mode']) ? (string)$stLocal['mode'] : 'local';
    $pid = isset($stLocal['pid']) ? (int)$stLocal['pid'] : 0;

    if ($mode === 'agent') {
      $respStop = agent_call('/stop', array('channel'=>array('id'=>$cid)));
      if (!isset($respStop['ok']) || !$respStop['ok']) {
        $err = isset($respStop['error']) ? (string)$respStop['error'] : 'agent stop failed';
        json_out(502, array('ok'=>false,'error'=>$err,'id'=>$cid,'mode'=>'agent'));
      }
    } else {
      if ($pid > 1) local_stop_ffmpeg($pid);
    }

    $stLocal['state'] = 'IDLE';
    $stLocal['pid'] = 0;
    $stLocal['stopped_at'] = date('c');
    save_state($stateKey, $stLocal);

    json_out(200, array('ok'=>true,'id'=>$cid,'state'=>'IDLE','mode'=>$mode));
  }

  if ($action === 'stop_project') {
    if ($userId === null || trim((string)$userId) === '') json_out(400, array('ok'=>false,'error'=>'missing user_id'));
    if (!$pdo) json_out(500, array('ok'=>false,'error'=>'db not configured'));
    $st = $pdo->prepare("SELECT channel_id FROM stream_channels WHERE user_id=:uid AND project=:p LIMIT 200");
    $st->execute(array(':uid'=>$userId, ':p'=>$project));
    $ids = array();
    foreach ($st->fetchAll() as $r) $ids[] = safe_id(isset($r['channel_id']) ? $r['channel_id'] : '');
    $stopped = array();

    foreach ($ids as $cid) {
      if ($cid === '') continue;
      $stateKey = state_key($project, $userId, $cid);
      $stLocal = load_state($stateKey);
      $mode = isset($stLocal['mode']) ? (string)$stLocal['mode'] : 'local';
      $pid = isset($stLocal['pid']) ? (int)$stLocal['pid'] : 0;
      if ($mode === 'agent') {
        try {
          $respStop = agent_call('/stop', array('channel'=>array('id'=>$cid)));
          if (!isset($respStop['ok']) || !$respStop['ok']) {
            $stopped[] = array('id'=>$cid,'ok'=>false,'error'=> (isset($respStop['error'])?(string)$respStop['error']:'agent stop failed'));
            continue;
          }
        } catch (Exception $e) {
          $stopped[] = array('id'=>$cid,'ok'=>false,'error'=>$e->getMessage());
          continue;
        }
      }
      else { if ($pid > 1) local_stop_ffmpeg($pid); }
      $stLocal['state']='IDLE'; $stLocal['pid']=0; $stLocal['stopped_at']=date('c');
      save_state($stateKey, $stLocal);
      $stopped[] = $cid;
    }
    json_out(200, array('ok'=>true,'stopped'=>$stopped));
  }

  if ($action === 'status') {
    $ids = array();
    if (isset($payload['channels']) && is_array($payload['channels'])) {
      foreach ($payload['channels'] as $c) {
        if (is_array($c) && isset($c['id'])) $ids[] = safe_id((string)$c['id']);
        else if (is_string($c)) $ids[] = safe_id($c);
      }
    }
    if (empty($ids) && $pdo && $userId !== null && trim((string)$userId) !== '') {
      $st = $pdo->prepare("SELECT channel_id FROM stream_channels WHERE user_id=:uid AND project=:p LIMIT 200");
      $st->execute(array(':uid'=>$userId, ':p'=>$project));
      foreach ($st->fetchAll() as $r) $ids[] = safe_id(isset($r['channel_id']) ? $r['channel_id'] : '');
    }

    $out = array();
    $uniq = array();
    foreach ($ids as $cid) { if ($cid !== '') $uniq[$cid] = true; }
    foreach (array_keys($uniq) as $cid) {
      $stateKey = state_key($project, $userId, $cid);
      $stLocal = load_state($stateKey);
      $mode = isset($stLocal['mode']) ? (string)$stLocal['mode'] : (can_local_spawn() ? 'local' : 'agent');

      if ($mode === 'agent') {
        try {
          $resp = agent_call('/status', array('channels'=>array(array('id'=>$cid))));
          if (isset($resp['channels']) && is_array($resp['channels']) && isset($resp['channels'][0]) && is_array($resp['channels'][0])) {
            $ch = $resp['channels'][0];
            $out[] = array(
              'id'=>$cid,
              'state'=>isset($ch['state']) ? (string)$ch['state'] : 'IDLE',
              'pid'=>isset($ch['pid']) ? (int)$ch['pid'] : 0,
              'mode'=>'agent',
              'container'=>isset($ch['container']) ? (string)$ch['container'] : '',
              'docker_status'=>isset($ch['docker_status']) ? (string)$ch['docker_status'] : '',
            );
            continue;
          }
        } catch (Exception $e) {
          // fall through
        }
      }

      $pid = isset($stLocal['pid']) ? (int)$stLocal['pid'] : 0;
      $running = ($pid > 1 && is_proc_running($pid));
      $out[] = array(
        'id'=>$cid,
        'state'=>$running ? 'RUNNING' : 'IDLE',
        'pid'=>$running ? $pid : 0,
        'mode'=>'local',
        'log'=>isset($stLocal['log']) ? (string)$stLocal['log'] : '',
      );
    }

    json_out(200, array('ok'=>true,'project'=>$project,'user_id'=>$userId,'channels'=>$out));
  }

  json_out(404, array('ok'=>false,'error'=>'unknown action','action'=>$action));

} catch (Exception $e) {
  json_out(500, array('ok'=>false,'error'=>$e->getMessage()));
}
