<?php
declare(strict_types=1);

final class LiveCastRepository {
    private static bool $schemaEnsured = false;

    public static function ensureSchema(): void {
        if (self::$schemaEnsured) {
            return;
        }
        $pdo = Database::pdo();
        Database::withMigrationLock($pdo, 'live_cast_sessions', static function () use ($pdo): void {
            self::createTables($pdo);
            self::ensureIndexes($pdo);
        });
        self::$schemaEnsured = true;
    }

    private static function createTables(PDO $pdo): void {
        if (!Database::tableExists($pdo, 'live_cast_sessions')) {
            if (Database::isMysql()) {
                $pdo->exec(
                    "CREATE TABLE live_cast_sessions (
                        id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
                        session_id VARCHAR(96) NOT NULL,
                        company_id INT UNSIGNED NOT NULL DEFAULT 1,
                        owner_user_id INT UNSIGNED NULL,
                        owner_username VARCHAR(64) NULL,
                        device_id VARCHAR(128) NULL,
                        platform VARCHAR(24) NOT NULL DEFAULT 'android',
                        status ENUM('intent','live','closed','expired') NOT NULL DEFAULT 'intent',
                        started_at_ms BIGINT NOT NULL,
                        last_heartbeat_ms BIGINT NOT NULL,
                        ended_at_ms BIGINT NULL,
                        max_duration_s INT NOT NULL DEFAULT 300,
                        join_short_url TEXT NULL,
                        join_qr_url TEXT NULL,
                        join_slug VARCHAR(32) NULL,
                        direct_short_url TEXT NULL,
                        direct_qr_url TEXT NULL,
                        direct_slug VARCHAR(32) NULL,
                        codec_video VARCHAR(32) NULL,
                        codec_audio VARCHAR(32) NULL,
                        bw_mode VARCHAR(24) NULL,
                        metadata_json LONGTEXT NULL,
                        created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                        updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                        UNIQUE KEY uq_live_cast_session_id (session_id),
                        KEY idx_live_cast_company_status (company_id, status),
                        KEY idx_live_cast_last_hb (last_heartbeat_ms),
                        KEY idx_live_cast_owner (owner_user_id)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
                );
            } else {
                $pdo->exec(
                    "CREATE TABLE live_cast_sessions (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        session_id TEXT NOT NULL UNIQUE,
                        company_id INTEGER NOT NULL DEFAULT 1,
                        owner_user_id INTEGER NULL,
                        owner_username TEXT NULL,
                        device_id TEXT NULL,
                        platform TEXT NOT NULL DEFAULT 'android',
                        status TEXT NOT NULL DEFAULT 'intent',
                        started_at_ms INTEGER NOT NULL,
                        last_heartbeat_ms INTEGER NOT NULL,
                        ended_at_ms INTEGER NULL,
                        max_duration_s INTEGER NOT NULL DEFAULT 300,
                        join_short_url TEXT NULL,
                        join_qr_url TEXT NULL,
                        join_slug TEXT NULL,
                        direct_short_url TEXT NULL,
                        direct_qr_url TEXT NULL,
                        direct_slug TEXT NULL,
                        codec_video TEXT NULL,
                        codec_audio TEXT NULL,
                        bw_mode TEXT NULL,
                        metadata_json TEXT NULL,
                        created_at TEXT NOT NULL DEFAULT (datetime('now')),
                        updated_at TEXT NOT NULL DEFAULT (datetime('now'))
                    )"
                );
            }
        }

        if (!Database::tableExists($pdo, 'live_cast_join_events')) {
            if (Database::isMysql()) {
                $pdo->exec(
                    "CREATE TABLE live_cast_join_events (
                        id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
                        session_id VARCHAR(96) NOT NULL,
                        company_id INT UNSIGNED NOT NULL DEFAULT 1,
                        event_type ENUM('open','join','leave') NOT NULL DEFAULT 'open',
                        platform VARCHAR(24) NOT NULL DEFAULT 'unknown',
                        channel VARCHAR(24) NOT NULL DEFAULT 'short_link',
                        direct TINYINT(1) NOT NULL DEFAULT 0,
                        user_id INT UNSIGNED NULL,
                        anon_id VARCHAR(96) NULL,
                        user_agent VARCHAR(255) NULL,
                        ip_hash CHAR(64) NULL,
                        created_at_ms BIGINT NOT NULL,
                        created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                        KEY idx_live_join_session_time (session_id, created_at_ms),
                        KEY idx_live_join_company_time (company_id, created_at_ms)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
                );
            } else {
                $pdo->exec(
                    "CREATE TABLE live_cast_join_events (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        session_id TEXT NOT NULL,
                        company_id INTEGER NOT NULL DEFAULT 1,
                        event_type TEXT NOT NULL DEFAULT 'open',
                        platform TEXT NOT NULL DEFAULT 'unknown',
                        channel TEXT NOT NULL DEFAULT 'short_link',
                        direct INTEGER NOT NULL DEFAULT 0,
                        user_id INTEGER NULL,
                        anon_id TEXT NULL,
                        user_agent TEXT NULL,
                        ip_hash TEXT NULL,
                        created_at_ms INTEGER NOT NULL,
                        created_at TEXT NOT NULL DEFAULT (datetime('now'))
                    )"
                );
            }
        }
    }

    private static function ensureIndexes(PDO $pdo): void {
        if (Database::isMysql()) {
            return;
        }
        $pdo->exec('CREATE INDEX IF NOT EXISTS idx_live_cast_company_status ON live_cast_sessions (company_id, status)');
        $pdo->exec('CREATE INDEX IF NOT EXISTS idx_live_cast_last_hb ON live_cast_sessions (last_heartbeat_ms)');
        $pdo->exec('CREATE INDEX IF NOT EXISTS idx_live_join_session_time ON live_cast_join_events (session_id, created_at_ms)');
        $pdo->exec('CREATE INDEX IF NOT EXISTS idx_live_join_company_time ON live_cast_join_events (company_id, created_at_ms)');
    }

    /** @return array<string,mixed> */
    public static function createIntent(array $input, array $actor): array {
        self::ensureSchema();
        self::expireStale();
        $pdo = Database::pdo();
        $now = (int) floor(microtime(true) * 1000);
        $userId = (int) ($actor['id'] ?? 0);
        $companyId = Rbac::activeCompanyId($actor) ?? Rbac::defaultCompanyId();
        $ownerUsername = trim((string) ($actor['username'] ?? ''));
        $platform = strtolower(trim((string) ($input['platform'] ?? 'android')));
        if ($platform === '') {
            $platform = 'android';
        }
        $deviceId = trim((string) ($input['device_id'] ?? ''));
        $maxDuration = (int) ($input['max_duration_s'] ?? 300);
        $maxDuration = max(60, min(1800, $maxDuration));
        if (str_starts_with($platform, 'education')) {
            $maxDuration = min($maxDuration, 300);
        }

        $sessionId = self::newSessionId($now, $companyId, $userId);
        $joinLongUrl = self::joinLongUrl($sessionId, false);
        $directLongUrl = self::joinLongUrl($sessionId, true);

        $links = self::makeShortLinks($joinLongUrl, $directLongUrl);
        $meta = is_array($input['meta'] ?? null) ? $input['meta'] : [];

        $stmt = $pdo->prepare(
            'INSERT INTO live_cast_sessions (
                session_id, company_id, owner_user_id, owner_username, device_id, platform, status,
                started_at_ms, last_heartbeat_ms, max_duration_s,
                join_short_url, join_qr_url, join_slug, direct_short_url, direct_qr_url, direct_slug,
                codec_video, codec_audio, bw_mode, metadata_json
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
        );
        $stmt->execute([
            $sessionId,
            $companyId,
            $userId > 0 ? $userId : null,
            $ownerUsername !== '' ? $ownerUsername : null,
            $deviceId !== '' ? $deviceId : null,
            $platform,
            'intent',
            $now,
            $now,
            $maxDuration,
            $links['join_short_url'],
            $links['join_qr_url'],
            $links['join_slug'],
            $links['direct_short_url'],
            $links['direct_qr_url'],
            $links['direct_slug'],
            trim((string) ($input['codec_video'] ?? '')) ?: null,
            trim((string) ($input['codec_audio'] ?? '')) ?: null,
            trim((string) ($input['bw_mode'] ?? '')) ?: null,
            safe_json_encode($meta, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE),
        ]);

        return self::sessionById($sessionId) ?? [];
    }

    /** @return array<string,mixed>|null */
    public static function heartbeat(string $sessionId, array $input, array $actor): ?array {
        self::ensureSchema();
        self::expireStale();
        $sessionId = trim($sessionId);
        if ($sessionId === '') {
            return null;
        }
        $pdo = Database::pdo();
        $now = (int) floor(microtime(true) * 1000);
        $companyId = Rbac::activeCompanyId($actor) ?? Rbac::defaultCompanyId();
        $status = strtolower(trim((string) ($input['status'] ?? 'active')));
        if ($status !== 'ended' && $status !== 'active') {
            $status = 'active';
        }
        $nextState = $status === 'ended' ? 'closed' : 'live';
        $stmt = $pdo->prepare(
            'UPDATE live_cast_sessions
             SET status = ?, last_heartbeat_ms = ?, ended_at_ms = CASE WHEN ? = ? THEN ? ELSE ended_at_ms END,
                 codec_video = COALESCE(NULLIF(?, \'\'), codec_video),
                 codec_audio = COALESCE(NULLIF(?, \'\'), codec_audio),
                 bw_mode = COALESCE(NULLIF(?, \'\'), bw_mode)
             WHERE session_id = ? AND company_id = ?'
        );
        $stmt->execute([
            $nextState,
            $now,
            $nextState,
            'closed',
            $now,
            trim((string) ($input['codec_video'] ?? '')),
            trim((string) ($input['codec_audio'] ?? '')),
            trim((string) ($input['bw_mode'] ?? '')),
            $sessionId,
            $companyId,
        ]);
        if ($stmt->rowCount() < 1) {
            return null;
        }
        if (isset($input['stats']) && is_array($input['stats'])) {
            $row = self::sessionById($sessionId);
            $meta = [];
            if ($row !== null && !empty($row['metadata_json'])) {
                $decoded = json_decode((string) $row['metadata_json'], true);
                if (is_array($decoded)) {
                    $meta = $decoded;
                }
            }
            $meta['last_stats'] = $input['stats'];
            $meta['last_stats_ms'] = $now;
            $upd = $pdo->prepare('UPDATE live_cast_sessions SET metadata_json = ? WHERE session_id = ?');
            $upd->execute([
                safe_json_encode($meta, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE),
                $sessionId,
            ]);
        }
        return self::sessionById($sessionId);
    }

    /** @return list<array<string,mixed>> */
    public static function activeSessions(array $actor, ?string $platform = null): array {
        self::ensureSchema();
        self::expireStale();
        $pdo = Database::pdo();
        $companyId = Rbac::activeCompanyId($actor) ?? Rbac::defaultCompanyId();
        $where = ['company_id = ?', "status IN ('intent','live')"];
        $params = [$companyId];
        if ($platform !== null && trim($platform) !== '') {
            $where[] = 'platform = ?';
            $params[] = strtolower(trim($platform));
        }
        $sql = 'SELECT session_id, owner_user_id, owner_username, device_id, platform, status,
                       started_at_ms, last_heartbeat_ms, max_duration_s, ended_at_ms,
                       join_short_url, join_qr_url, direct_short_url, direct_qr_url,
                       codec_video, codec_audio, bw_mode
                FROM live_cast_sessions
                WHERE ' . implode(' AND ', $where) . '
                ORDER BY last_heartbeat_ms DESC LIMIT 200';
        $stmt = $pdo->prepare($sql);
        $stmt->execute($params);
        $out = [];
        foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $out[] = self::normalizeSessionRow($row);
        }
        return $out;
    }

    /** @return array<string,mixed> */
    public static function analytics(int $days, array $actor): array {
        self::ensureSchema();
        self::expireStale();
        $days = max(1, min(90, $days));
        $startMs = (int) ((time() - ($days * 86400)) * 1000);
        $companyId = Rbac::activeCompanyId($actor) ?? Rbac::defaultCompanyId();
        $pdo = Database::pdo();

        $castsPerDay = self::countByDay($pdo, 'live_cast_sessions', 'started_at_ms', $startMs, $companyId);
        $avgDurationPerDay = self::avgDurationByDay($pdo, $startMs, $companyId);
        $platformMix = self::breakdown($pdo, 'live_cast_sessions', 'platform', $startMs, $companyId);
        $videoCodecMix = self::breakdown($pdo, 'live_cast_sessions', 'codec_video', $startMs, $companyId);
        $audioCodecMix = self::breakdown($pdo, 'live_cast_sessions', 'codec_audio', $startMs, $companyId);
        $bwModeMix = self::breakdown($pdo, 'live_cast_sessions', 'bw_mode', $startMs, $companyId);
        $joinsPerPlatform = self::breakdown($pdo, 'live_cast_join_events', 'platform', $startMs, $companyId);

        $topUsersStmt = $pdo->prepare(
            'SELECT COALESCE(NULLIF(owner_username, \'\'), \'unknown\') AS u, COUNT(*) AS c
             FROM live_cast_sessions
             WHERE company_id = ? AND started_at_ms >= ?
             GROUP BY u ORDER BY c DESC LIMIT 20'
        );
        $topUsersStmt->execute([$companyId, $startMs]);
        $topUsers = [];
        foreach ($topUsersStmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $topUsers[] = ['label' => (string) ($row['u'] ?? 'unknown'), 'value' => (int) ($row['c'] ?? 0)];
        }

        return [
            'window_days' => $days,
            'casts_per_day' => $castsPerDay,
            'avg_length_s_per_day' => $avgDurationPerDay,
            'platform_mix' => $platformMix,
            'joins_platform_mix' => $joinsPerPlatform,
            'video_codec_mix' => $videoCodecMix,
            'audio_codec_mix' => $audioCodecMix,
            'bw_mode_mix' => $bwModeMix,
            'top_users' => $topUsers,
        ];
    }

    public static function registerJoin(string $sessionId, array $payload, array $actor): bool {
        self::ensureSchema();
        $sessionId = trim($sessionId);
        if ($sessionId === '') {
            return false;
        }
        $session = self::sessionById($sessionId);
        if ($session === null) {
            return false;
        }
        $pdo = Database::pdo();
        $now = (int) floor(microtime(true) * 1000);
        $companyId = (int) ($session['company_id'] ?? Rbac::defaultCompanyId());
        $eventType = strtolower(trim((string) ($payload['event_type'] ?? 'open')));
        if (!in_array($eventType, ['open', 'join', 'leave'], true)) {
            $eventType = 'open';
        }
        $platform = strtolower(trim((string) ($payload['platform'] ?? 'unknown')));
        if ($platform === '') {
            $platform = 'unknown';
        }
        $channel = strtolower(trim((string) ($payload['channel'] ?? 'short_link')));
        if ($channel === '') {
            $channel = 'short_link';
        }
        $direct = !empty($payload['direct']) ? 1 : 0;
        $userId = (int) ($actor['id'] ?? 0);
        $anon = trim((string) ($payload['anon_id'] ?? ''));
        $ua = trim((string) ($_SERVER['HTTP_USER_AGENT'] ?? ''));
        $ip = (string) ($_SERVER['HTTP_X_FORWARDED_FOR'] ?? ($_SERVER['REMOTE_ADDR'] ?? ''));
        $ipHash = $ip !== '' ? hash('sha256', $ip . '|ac-live-cast') : null;

        $stmt = $pdo->prepare(
            'INSERT INTO live_cast_join_events
             (session_id, company_id, event_type, platform, channel, direct, user_id, anon_id, user_agent, ip_hash, created_at_ms)
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
        );
        $stmt->execute([
            $sessionId,
            $companyId,
            $eventType,
            $platform,
            $channel,
            $direct,
            $userId > 0 ? $userId : null,
            $anon !== '' ? $anon : null,
            $ua !== '' ? substr($ua, 0, 255) : null,
            $ipHash,
            $now,
        ]);
        return true;
    }

    /** @return array<string,mixed>|null */
    public static function sessionById(string $sessionId): ?array {
        self::ensureSchema();
        $stmt = Database::pdo()->prepare(
            'SELECT * FROM live_cast_sessions WHERE session_id = ? LIMIT 1'
        );
        $stmt->execute([$sessionId]);
        $row = $stmt->fetch(PDO::FETCH_ASSOC);
        if (!is_array($row)) {
            return null;
        }
        return self::normalizeSessionRow($row);
    }

    private static function expireStale(): void {
        $pdo = Database::pdo();
        $now = (int) floor(microtime(true) * 1000);
        $staleBefore = $now - 60_000;
        $sql = "UPDATE live_cast_sessions
                SET status = 'expired', ended_at_ms = COALESCE(ended_at_ms, ?)
                WHERE status IN ('intent','live')
                  AND last_heartbeat_ms < ?";
        $stmt = $pdo->prepare($sql);
        $stmt->execute([$now, $staleBefore]);
    }

    /** @return array<string,mixed> */
    private static function normalizeSessionRow(array $row): array {
        return [
            'session_id' => (string) ($row['session_id'] ?? ''),
            'company_id' => (int) ($row['company_id'] ?? 0),
            'owner_user_id' => isset($row['owner_user_id']) ? (int) $row['owner_user_id'] : null,
            'owner_username' => (string) ($row['owner_username'] ?? ''),
            'device_id' => (string) ($row['device_id'] ?? ''),
            'platform' => (string) ($row['platform'] ?? ''),
            'status' => (string) ($row['status'] ?? ''),
            'started_at_ms' => (int) ($row['started_at_ms'] ?? 0),
            'last_heartbeat_ms' => (int) ($row['last_heartbeat_ms'] ?? 0),
            'ended_at_ms' => isset($row['ended_at_ms']) ? (int) $row['ended_at_ms'] : null,
            'max_duration_s' => (int) ($row['max_duration_s'] ?? 0),
            'join_short_url' => (string) ($row['join_short_url'] ?? ''),
            'join_qr_url' => (string) ($row['join_qr_url'] ?? ''),
            'direct_short_url' => (string) ($row['direct_short_url'] ?? ''),
            'direct_qr_url' => (string) ($row['direct_qr_url'] ?? ''),
            'codec_video' => (string) ($row['codec_video'] ?? ''),
            'codec_audio' => (string) ($row['codec_audio'] ?? ''),
            'bw_mode' => (string) ($row['bw_mode'] ?? ''),
            'metadata_json' => (string) ($row['metadata_json'] ?? ''),
        ];
    }

    private static function newSessionId(int $nowMs, int $companyId, int $userId): string {
        $seed = $nowMs . ':' . $companyId . ':' . $userId . ':' . bin2hex(random_bytes(6));
        return 'live-' . substr(hash('sha256', $seed), 0, 20);
    }

    private static function joinLongUrl(string $sessionId, bool $direct): string {
        $base = rtrim((string) cfg('base_path', '/app/androidcast_project/issues'), '/');
        if (str_ends_with($base, '/crashes')) {
            $base = substr($base, 0, -strlen('/crashes')) . '/issues';
        }
        $qs = 'session_id=' . rawurlencode($sessionId);
        if ($direct) {
            $qs .= '&direct=1';
        }
        return 'https://apps.f0xx.org' . $base . '/live/join?' . $qs;
    }

    /** @return array<string, string|null> */
    private static function makeShortLinks(string $joinLongUrl, string $directLongUrl): array {
        $out = [
            'join_short_url' => null,
            'join_qr_url' => null,
            'join_slug' => null,
            'direct_short_url' => null,
            'direct_qr_url' => null,
            'direct_slug' => null,
        ];
        if (!UrlShortenerDatabase::enabled() || !UrlShortenerDatabase::ping()) {
            return $out;
        }
        $bearers = ShortLinksRepository::listBearers();
        if ($bearers === []) {
            return $out;
        }
        $bearerId = (int) ($bearers[0]['id'] ?? 0);
        if ($bearerId <= 0) {
            return $out;
        }
        $join = ShortLinksRepository::createLink($bearerId, $joinLongUrl, 3600);
        if (!empty($join['ok'])) {
            $out['join_short_url'] = (string) ($join['short_url'] ?? '');
            $out['join_qr_url'] = (string) ($join['qr_url'] ?? '');
            $out['join_slug'] = (string) ($join['slug'] ?? '');
        }
        $direct = ShortLinksRepository::createLink($bearerId, $directLongUrl, 3600);
        if (!empty($direct['ok'])) {
            $out['direct_short_url'] = (string) ($direct['short_url'] ?? '');
            $out['direct_qr_url'] = (string) ($direct['qr_url'] ?? '');
            $out['direct_slug'] = (string) ($direct['slug'] ?? '');
        }
        return $out;
    }

    /** @return list<array{x:string,y:int}> */
    private static function countByDay(PDO $pdo, string $table, string $msColumn, int $startMs, int $companyId): array {
        $day = Database::isMysql()
            ? 'DATE(FROM_UNIXTIME(FLOOR(' . $msColumn . ' / 1000)))'
            : "strftime('%Y-%m-%d', " . $msColumn . " / 1000, 'unixepoch')";
        $stmt = $pdo->prepare(
            'SELECT ' . $day . ' AS d, COUNT(*) AS c FROM ' . $table . '
             WHERE company_id = ? AND ' . $msColumn . ' >= ?
             GROUP BY d ORDER BY d'
        );
        $stmt->execute([$companyId, $startMs]);
        $out = [];
        foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $out[] = ['x' => (string) ($row['d'] ?? ''), 'y' => (int) ($row['c'] ?? 0)];
        }
        return $out;
    }

    /** @return list<array{x:string,y:float}> */
    private static function avgDurationByDay(PDO $pdo, int $startMs, int $companyId): array {
        $day = Database::isMysql()
            ? 'DATE(FROM_UNIXTIME(FLOOR(started_at_ms / 1000)))'
            : "strftime('%Y-%m-%d', started_at_ms / 1000, 'unixepoch')";
        $stmt = $pdo->prepare(
            'SELECT ' . $day . ' AS d,
                    AVG(CASE WHEN ended_at_ms IS NOT NULL AND ended_at_ms >= started_at_ms
                             THEN (ended_at_ms - started_at_ms) / 1000.0 ELSE 0 END) AS v
             FROM live_cast_sessions
             WHERE company_id = ? AND started_at_ms >= ?
             GROUP BY d ORDER BY d'
        );
        $stmt->execute([$companyId, $startMs]);
        $out = [];
        foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $out[] = ['x' => (string) ($row['d'] ?? ''), 'y' => round((float) ($row['v'] ?? 0), 1)];
        }
        return $out;
    }

    /** @return list<array<string,mixed>> */
    public static function listSessions(int $days, array $actor, int $limit = 200): array {
        self::ensureSchema();
        self::expireStale();
        $days = max(1, min(90, $days));
        $startMs = (int) ((time() - ($days * 86400)) * 1000);
        $companyId = Rbac::activeCompanyId($actor) ?? Rbac::defaultCompanyId();
        $limit = max(1, min(500, $limit));
        $pdo = Database::pdo();
        $stmt = $pdo->prepare(
            'SELECT s.session_id, s.owner_user_id, s.owner_username, s.device_id, s.platform, s.status,
                    s.started_at_ms, s.last_heartbeat_ms, s.ended_at_ms, s.max_duration_s,
                    s.join_short_url, s.direct_short_url, s.codec_video, s.codec_audio, s.bw_mode,
                    (SELECT COUNT(*) FROM live_cast_join_events j
                     WHERE j.session_id = s.session_id AND j.event_type IN (\'open\', \'join\')) AS join_opens
             FROM live_cast_sessions s
             WHERE s.company_id = ? AND s.started_at_ms >= ?
             ORDER BY s.last_heartbeat_ms DESC
             LIMIT ' . (int) $limit
        );
        $stmt->execute([$companyId, $startMs]);
        $out = [];
        foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $normalized = self::normalizeSessionRow($row);
            $normalized['join_opens'] = (int) ($row['join_opens'] ?? 0);
            $out[] = $normalized;
        }
        return $out;
    }

    /** @return array<string,mixed>|null */
    public static function publicSession(string $sessionId): ?array {
        $row = self::sessionById($sessionId);
        if ($row === null) {
            return null;
        }
        return [
            'session_id' => $row['session_id'],
            'owner_username' => $row['owner_username'],
            'platform' => $row['platform'],
            'status' => $row['status'],
            'started_at_ms' => $row['started_at_ms'],
            'last_heartbeat_ms' => $row['last_heartbeat_ms'],
            'ended_at_ms' => $row['ended_at_ms'],
            'max_duration_s' => $row['max_duration_s'],
            'codec_video' => $row['codec_video'],
            'codec_audio' => $row['codec_audio'],
            'bw_mode' => $row['bw_mode'],
            'join_short_url' => $row['join_short_url'] ?? '',
            'join_qr_url' => $row['join_qr_url'] ?? '',
            'signaling_api' => Auth::basePath() . '/api/live_cast_signaling.php',
        ];
    }

    /** @return list<array{label:string,value:int}> */
    private static function breakdown(PDO $pdo, string $table, string $column, int $startMs, int $companyId): array {
        $allowedTables = ['live_cast_sessions', 'live_cast_join_events'];
        if (!in_array($table, $allowedTables, true)) {
            return [];
        }
        $allowedCols = ['platform', 'codec_video', 'codec_audio', 'bw_mode'];
        if ($table === 'live_cast_join_events') {
            $allowedCols = ['platform'];
        }
        if (!in_array($column, $allowedCols, true)) {
            return [];
        }
        $timeColumn = $table === 'live_cast_sessions' ? 'started_at_ms' : 'created_at_ms';
        $stmt = $pdo->prepare(
            'SELECT COALESCE(NULLIF(' . $column . ', \'\'), \'unknown\') AS k, COUNT(*) AS c
             FROM ' . $table . '
             WHERE company_id = ? AND ' . $timeColumn . ' >= ?
             GROUP BY k ORDER BY c DESC'
        );
        $stmt->execute([$companyId, $startMs]);
        $out = [];
        foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
            $out[] = ['label' => (string) ($row['k'] ?? 'unknown'), 'value' => (int) ($row['c'] ?? 0)];
        }
        return $out;
    }
}

