From 224dc80d6d2adbc00b14af13b20072556a2f8093 Mon Sep 17 00:00:00 2001 From: Anton Afanasyeu Date: Tue, 23 Jun 2026 12:29:33 +0200 Subject: [PATCH] initial --- README.md | 3 + composer.json | 25 + config/config.example.php | 2 + public/api/graph_upload.php | 32 + public/api/graphs.php | 22 + public/index.php | 2 + sql/migrations/006_graph_sessions.sql | 30 + sql/migrations/006_graph_sessions.sqlite.sql | 6 + sql/migrations/008_graph_sessions_indexes.sql | 5 + src/GraphRepository.php | 600 ++++++++++++++++++ 10 files changed, 727 insertions(+) create mode 100644 README.md create mode 100644 composer.json create mode 100644 config/config.example.php create mode 100644 public/api/graph_upload.php create mode 100644 public/api/graphs.php create mode 100644 public/index.php create mode 100644 sql/migrations/006_graph_sessions.sql create mode 100644 sql/migrations/006_graph_sessions.sqlite.sql create mode 100644 sql/migrations/008_graph_sessions_indexes.sql create mode 100644 src/GraphRepository.php diff --git a/README.md b/README.md new file mode 100644 index 0000000..7c26861 --- /dev/null +++ b/README.md @@ -0,0 +1,3 @@ +# ac-ms-graphs + +Microservice API. Remote: `git://f0xx.org/ac/ac-ms-graphs` diff --git a/composer.json b/composer.json new file mode 100644 index 0000000..c799188 --- /dev/null +++ b/composer.json @@ -0,0 +1,25 @@ +{ + "name": "androidcast/ms-graphs", + "description": "Graph session ingest/query", + "type": "project", + "require": { + "php": ">=8.1", + "androidcast/platform-php": "dev-next", + "androidcast/platform-db": "dev-next" + }, + "repositories": [ + { + "type": "vcs", + "url": "git://f0xx.org/ac/ac-platform-php" + }, + { + "type": "vcs", + "url": "git://f0xx.org/ac/ac-platform-db" + } + ], + "autoload": { + "classmap": [ + "src/" + ] + } +} diff --git a/config/config.example.php b/config/config.example.php new file mode 100644 index 0000000..9a59c16 --- /dev/null +++ b/config/config.example.php @@ -0,0 +1,2 @@ + ['driver' => 'sqlite']]; diff --git a/public/api/graph_upload.php b/public/api/graph_upload.php new file mode 100644 index 0000000..e686395 --- /dev/null +++ b/public/api/graph_upload.php @@ -0,0 +1,32 @@ + false, 'error' => 'empty body'], 400); +} +$data = json_decode($raw, true); +if (!is_array($data)) { + json_out(['ok' => false, 'error' => 'invalid json'], 400); +} +if ((int) ($data['schema_version'] ?? 0) !== 1) { + json_out(['ok' => false, 'error' => 'unsupported schema_version'], 400); +} +if (($data['ingest_kind'] ?? 'graph_session') !== 'graph_session') { + json_out(['ok' => false, 'error' => 'ingest_kind must be graph_session'], 400); +} + +try { + GraphRepository::insertSession($data); + json_out(['ok' => true, 'session_id' => $data['session_id'] ?? null]); +} catch (InvalidArgumentException $e) { + json_out(['ok' => false, 'error' => $e->getMessage()], 400); +} catch (Throwable $e) { + error_log('graph upload: ' . $e->getMessage()); + $out = ['ok' => false, 'error' => 'store_failed']; + if (cfg('debug')) { + $out['hint'] = $e->getMessage(); + } + json_out($out, 500); +} diff --git a/public/api/graphs.php b/public/api/graphs.php new file mode 100644 index 0000000..fe34066 --- /dev/null +++ b/public/api/graphs.php @@ -0,0 +1,22 @@ + true, 'data' => $payload]); +} catch (Throwable $e) { + error_log('graphs api: ' . $e->getMessage()); + $out = ['ok' => false, 'error' => 'graphs_load_failed']; + if (cfg('debug')) { + $out['hint'] = $e->getMessage(); + } + json_out($out, 500); +} diff --git a/public/index.php b/public/index.php new file mode 100644 index 0000000..922b7c2 --- /dev/null +++ b/public/index.php @@ -0,0 +1,2 @@ +prepare( + 'SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ? LIMIT 1' + ); + $stmt->execute([$table, $column]); + return (bool) $stmt->fetchColumn(); + } + $stmt = $pdo->query('PRAGMA table_info(' . $table . ')'); + if ($stmt === false) { + return false; + } + while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) { + if (($row['name'] ?? '') === $column) { + return true; + } + } + return false; + } + + private static function indexExists(PDO $pdo, string $table, string $index): bool { + if (Database::isMysql()) { + $stmt = $pdo->prepare( + 'SELECT 1 FROM INFORMATION_SCHEMA.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ? LIMIT 1' + ); + $stmt->execute([$table, $index]); + return (bool) $stmt->fetchColumn(); + } + $stmt = $pdo->query('PRAGMA index_list(' . $table . ')'); + if ($stmt === false) { + return false; + } + while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) { + if (($row['name'] ?? '') === $index) { + return true; + } + } + return false; + } + + private static function createTable(PDO $pdo): void { + if (Database::isMysql()) { + $pdo->exec( + "CREATE TABLE IF NOT EXISTS graph_sessions ( + id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, + session_id VARCHAR(96) NOT NULL, + company_id INT UNSIGNED NOT NULL DEFAULT 1, + user_id INT UNSIGNED NULL, + device_id VARCHAR(128) NULL, + app_version VARCHAR(64) NULL, + sdk_int INT NULL, + started_at_ms BIGINT NOT NULL, + ended_at_ms BIGINT NOT NULL, + duration_s INT NOT NULL, + completed TINYINT(1) NOT NULL DEFAULT 1, + direction ENUM('sender','receiver') NOT NULL DEFAULT 'sender', + transport VARCHAR(24) NOT NULL DEFAULT 'wifi', + avg_kbps INT NOT NULL DEFAULT 0, + recv_kbps INT NOT NULL DEFAULT 0, + packet_loss_pct DECIMAL(6,3) NOT NULL DEFAULT 0.0, + ntp_source VARCHAR(32) NOT NULL DEFAULT 'device', + ntp_correction_s INT NULL, + install_source VARCHAR(32) NOT NULL DEFAULT 'direct', + meta_json LONGTEXT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE KEY uq_graph_session_id (session_id), + KEY idx_graph_company_time (company_id, started_at_ms), + KEY idx_graph_device (device_id), + KEY idx_graph_app_version (app_version) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci" + ); + return; + } + $pdo->exec( + "CREATE TABLE IF NOT EXISTS graph_sessions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL UNIQUE, + company_id INTEGER NOT NULL DEFAULT 1, + user_id INTEGER NULL, + device_id TEXT NULL, + app_version TEXT NULL, + sdk_int INTEGER NULL, + started_at_ms INTEGER NOT NULL, + ended_at_ms INTEGER NOT NULL, + duration_s INTEGER NOT NULL, + completed INTEGER NOT NULL DEFAULT 1, + direction TEXT NOT NULL DEFAULT 'sender', + transport TEXT NOT NULL DEFAULT 'wifi', + avg_kbps INTEGER NOT NULL DEFAULT 0, + recv_kbps INTEGER NOT NULL DEFAULT 0, + packet_loss_pct REAL NOT NULL DEFAULT 0.0, + ntp_source TEXT NOT NULL DEFAULT 'device', + ntp_correction_s INTEGER NULL, + install_source TEXT NOT NULL DEFAULT 'direct', + meta_json TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + )" + ); + } + + private static function ensureIndexes(PDO $pdo): void { + $defs = Database::isMysql() + ? [ + 'idx_graph_user_time' => 'CREATE INDEX idx_graph_user_time ON graph_sessions (user_id, started_at_ms)', + 'idx_graph_started' => 'CREATE INDEX idx_graph_started ON graph_sessions (started_at_ms)', + ] + : [ + 'idx_graph_company_time' => 'CREATE INDEX IF NOT EXISTS idx_graph_company_time ON graph_sessions (company_id, started_at_ms)', + 'idx_graph_user_time' => 'CREATE INDEX IF NOT EXISTS idx_graph_user_time ON graph_sessions (user_id, started_at_ms)', + 'idx_graph_started' => 'CREATE INDEX IF NOT EXISTS idx_graph_started ON graph_sessions (started_at_ms)', + 'idx_graph_device' => 'CREATE INDEX IF NOT EXISTS idx_graph_device ON graph_sessions (device_id)', + ]; + foreach ($defs as $name => $sql) { + if (self::indexExists($pdo, 'graph_sessions', $name)) { + continue; + } + try { + $pdo->exec($sql); + } catch (PDOException $e) { + $msg = strtolower($e->getMessage()); + if (!str_contains($msg, 'duplicate') && !str_contains($msg, 'already exists')) { + throw $e; + } + } + } + } + + private static function ensureColumns(PDO $pdo): void { + $cols = [ + 'device_id' => Database::isMysql() ? 'VARCHAR(128) NULL' : 'TEXT NULL', + 'app_version' => Database::isMysql() ? 'VARCHAR(64) NULL' : 'TEXT NULL', + 'sdk_int' => Database::isMysql() ? 'INT NULL' : 'INTEGER NULL', + 'completed' => Database::isMysql() ? 'TINYINT(1) NOT NULL DEFAULT 1' : 'INTEGER NOT NULL DEFAULT 1', + 'recv_kbps' => Database::isMysql() ? 'INT NOT NULL DEFAULT 0' : 'INTEGER NOT NULL DEFAULT 0', + 'ntp_correction_s' => Database::isMysql() ? 'INT NULL' : 'INTEGER NULL', + ]; + foreach ($cols as $name => $ddl) { + if (self::columnExists($pdo, 'graph_sessions', $name)) { + continue; + } + try { + $pdo->exec('ALTER TABLE graph_sessions ADD COLUMN ' . $name . ' ' . $ddl); + } catch (PDOException $e) { + $msg = strtolower($e->getMessage()); + if (!str_contains($msg, 'duplicate') && !str_contains($msg, 'already exists')) { + throw $e; + } + } + } + } + + public static function insertSession(array $in): void { + self::ensureSchema(); + $pdo = Database::pdo(); + $sessionId = trim((string) ($in['session_id'] ?? '')); + if ($sessionId === '') { + throw new InvalidArgumentException('session_id is required'); + } + $start = (int) ($in['started_at_epoch_ms'] ?? 0); + $end = (int) ($in['ended_at_epoch_ms'] ?? 0); + if ($start <= 0 || $end <= 0 || $end < $start) { + throw new InvalidArgumentException('invalid session timing'); + } + $dir = strtolower(trim((string) ($in['direction'] ?? 'sender'))); + if ($dir !== 'sender' && $dir !== 'receiver') { + $dir = 'sender'; + } + $transport = strtolower(trim((string) ($in['transport'] ?? 'wifi'))); + if ($transport === '') { + $transport = 'wifi'; + } + $ntp = strtolower(trim((string) ($in['ntp_source'] ?? ($in['time_source'] ?? 'device')))); + if ($ntp === '') { + $ntp = 'device'; + } + $install = strtolower(trim((string) ($in['install_source'] ?? 'direct'))); + if ($install === '') { + $install = 'direct'; + } + $duration = max(0, (int) ($in['duration_s'] ?? (int) (($end - $start) / 1000))); + $avgKbps = max(0, (int) ($in['avg_kbps'] ?? 0)); + $recvKbps = max(0, (int) ($in['recv_kbps'] ?? 0)); + $loss = max(0.0, (float) ($in['packet_loss_pct'] ?? 0.0)); + $companyId = (int) ($in['company_id'] ?? Rbac::defaultCompanyId()); + if ($companyId <= 0) { + $companyId = Rbac::defaultCompanyId(); + } + $userId = isset($in['user_id']) ? (int) $in['user_id'] : null; + if ($userId !== null && $userId <= 0) { + $userId = null; + } + $deviceId = trim((string) ($in['device_id'] ?? '')); + if ($deviceId === '' && is_array($in['device'] ?? null)) { + $deviceId = trim((string) (($in['device']['id'] ?? $in['device']['device_id'] ?? ''))); + } + $appVersion = trim((string) ($in['app_version'] ?? '')); + if ($appVersion === '' && is_array($in['app'] ?? null)) { + $appVersion = trim((string) ($in['app']['version_name'] ?? $in['app']['version'] ?? '')); + } + $sdkInt = isset($in['sdk_int']) ? (int) $in['sdk_int'] : null; + if ($sdkInt === null && is_array($in['device'] ?? null)) { + $sdkInt = isset($in['device']['sdk_int']) ? (int) $in['device']['sdk_int'] : null; + } + $completed = 1; + if (array_key_exists('completed', $in)) { + $completed = !empty($in['completed']) ? 1 : 0; + } + $ntpCorrection = isset($in['ntp_correction_s']) ? (int) $in['ntp_correction_s'] : null; + if ($ntpCorrection === null && isset($in['ntp_correction_seconds'])) { + $ntpCorrection = (int) $in['ntp_correction_seconds']; + } + $meta = $in['meta'] ?? null; + $metaJson = is_array($meta) ? safe_json_encode($meta, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE) : null; + + if (Database::isMysql()) { + $sql = 'INSERT INTO graph_sessions + (session_id, company_id, user_id, device_id, app_version, sdk_int, + started_at_ms, ended_at_ms, duration_s, completed, direction, transport, + avg_kbps, recv_kbps, packet_loss_pct, ntp_source, ntp_correction_s, install_source, meta_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + ended_at_ms = VALUES(ended_at_ms), + duration_s = VALUES(duration_s), + completed = VALUES(completed), + avg_kbps = VALUES(avg_kbps), + recv_kbps = VALUES(recv_kbps), + packet_loss_pct = VALUES(packet_loss_pct), + ntp_source = VALUES(ntp_source), + ntp_correction_s = VALUES(ntp_correction_s), + install_source = VALUES(install_source), + device_id = COALESCE(VALUES(device_id), device_id), + app_version = COALESCE(VALUES(app_version), app_version), + sdk_int = COALESCE(VALUES(sdk_int), sdk_int), + meta_json = VALUES(meta_json)'; + } else { + $sql = 'INSERT OR REPLACE INTO graph_sessions + (session_id, company_id, user_id, device_id, app_version, sdk_int, + started_at_ms, ended_at_ms, duration_s, completed, direction, transport, + avg_kbps, recv_kbps, packet_loss_pct, ntp_source, ntp_correction_s, install_source, meta_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'; + } + $stmt = $pdo->prepare($sql); + $stmt->execute([ + $sessionId, + $companyId, + $userId, + $deviceId !== '' ? $deviceId : null, + $appVersion !== '' ? $appVersion : null, + $sdkInt, + $start, + $end, + $duration, + $completed, + $dir, + $transport, + $avgKbps, + $recvKbps, + $loss, + $ntp, + $ntpCorrection, + $install, + $metaJson, + ]); + } + + public static function dashboardData(int $days = 14): array { + self::ensureSchema(); + $days = max(1, min(60, $days)); + $user = Auth::user(); + $uid = (int) ($user['id'] ?? 0); + $scopeUserId = $uid > 0 ? $uid : null; + $activeCompanyId = Rbac::activeCompanyId($user) ?? Rbac::defaultCompanyId(); + $viewer = self::viewerRole($user); + + $payload = [ + 'window_days' => $days, + 'viewer' => $viewer, + 'links' => self::serviceLinks(), + ]; + $payload['live_cast'] = LiveCastRepository::analytics($days, $user ?? []); + $payload['user'] = self::buildScope($days, $activeCompanyId, null); + if ($viewer === 'user') { + return $payload; + } + $payload['slug_admin'] = self::buildScope($days, $activeCompanyId, null); + if ($viewer !== 'platform_admin') { + return $payload; + } + $payload['platform_admin'] = self::buildScope($days, null, null, true); + return $payload; + } + + private static function viewerRole(?array $user): string { + if (Rbac::isGlobalAdmin($user)) { + return 'platform_admin'; + } + $companyId = Rbac::activeCompanyId($user) ?? Rbac::defaultCompanyId(); + $uid = (int) ($user['id'] ?? 0); + $role = ($companyId > 0 && $uid > 0) ? Rbac::companyRole($uid, $companyId) : null; + if ($role === 'owner' || $role === 'admin') { + return 'slug_admin'; + } + return 'user'; + } + + /** @return array */ + private static function serviceLinks(): array { + $prefix = '/app/androidcast_project'; + return [ + 'hub' => $prefix . '/', + 'graphs' => $prefix . '/graphs/', + 'issues' => $prefix . '/issues/?view=reports', + 'tickets' => $prefix . '/issues/?view=tickets', + 'build' => $prefix . '/build/', + 'git' => $prefix . '/git/', + ]; + } + + private static function buildScope(int $days, ?int $companyId, ?int $userId, bool $platform = false): array { + $startMs = (int) ((time() - ($days * 86400)) * 1000); + $scope = [ + 'sessions_per_day' => self::sessionsPerDay($startMs, $companyId, $userId), + 'unique_devices_per_day' => self::uniqueDevicesPerDay($startMs, $companyId, $userId), + 'avg_duration_s_per_day' => self::avgDurationPerDay($startMs, $companyId, $userId), + 'send_sessions_per_day' => self::sessionsPerDay($startMs, $companyId, $userId, 'sender'), + 'recv_sessions_per_day' => self::sessionsPerDay($startMs, $companyId, $userId, 'receiver'), + 'success_ratio_pct' => self::successRatio($startMs, $companyId, $userId), + 'avg_bitrate_kbps_per_day' => self::avgBitratePerDay($startMs, $companyId, $userId, 'avg_kbps'), + 'avg_recv_kbps_per_day' => self::avgBitratePerDay($startMs, $companyId, $userId, 'recv_kbps'), + 'crashes_per_day' => self::crashesPerDay($startMs, $companyId), + 'ntp_sources' => self::breakdown('ntp_source', $startMs, $companyId, $userId), + 'install_sources' => self::breakdown('install_source', $startMs, $companyId, $userId), + 'transport_mix' => self::breakdown('transport', $startMs, $companyId, $userId), + 'app_versions' => self::breakdown('app_version', $startMs, $companyId, $userId), + ]; + if (Database::tableExists(Database::pdo(), 'tickets')) { + $scope['tickets_per_day'] = self::ticketsPerDay($startMs, $companyId); + } + if ($platform) { + $scope['top_crash_fingerprints'] = self::topCrashFingerprints($startMs, 8); + $scope['crashes_by_app_version'] = self::crashesByAppVersion($startMs); + $scope['ntp_correction_avg_s'] = self::avgNtpCorrection($startMs, $companyId, $userId); + } + return $scope; + } + + private static function sessionsPerDay( + int $startMs, + ?int $companyId, + ?int $userId, + ?string $direction = null + ): array { + $where = ['started_at_ms >= ?']; + $params = [$startMs]; + if ($direction !== null) { + $where[] = 'direction = ?'; + $params[] = $direction; + } + self::appendScope($where, $params, $companyId, $userId); + return self::countByDay('graph_sessions', 'started_at_ms', $where, $params); + } + + private static function uniqueDevicesPerDay(int $startMs, ?int $companyId, ?int $userId): array { + $where = ['started_at_ms >= ?', "device_id IS NOT NULL", "device_id <> ''"]; + $params = [$startMs]; + self::appendScope($where, $params, $companyId, $userId); + $day = self::msDayExpr('started_at_ms'); + $sql = 'SELECT ' . $day . ' AS d, COUNT(DISTINCT device_id) AS c FROM graph_sessions WHERE ' + . implode(' AND ', $where) . ' GROUP BY d ORDER BY d'; + $stmt = Database::pdo()->prepare($sql); + $stmt->execute($params); + $bucket = []; + foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) { + $bucket[(string) ($row['d'] ?? '')] = (int) ($row['c'] ?? 0); + } + return self::assocToSeries($bucket); + } + + private static function avgDurationPerDay(int $startMs, ?int $companyId, ?int $userId): array { + return self::avgNumericPerDay($startMs, $companyId, $userId, 'duration_s'); + } + + private static function successRatio(int $startMs, ?int $companyId, ?int $userId): float { + $where = ['started_at_ms >= ?']; + $params = [$startMs]; + self::appendScope($where, $params, $companyId, $userId); + $sql = 'SELECT COUNT(*) AS total, + SUM(CASE WHEN completed = 1 THEN 1 ELSE 0 END) AS ok + FROM graph_sessions WHERE ' . implode(' AND ', $where); + $stmt = Database::pdo()->prepare($sql); + $stmt->execute($params); + $row = $stmt->fetch(PDO::FETCH_ASSOC) ?: []; + $total = (int) ($row['total'] ?? 0); + $ok = (int) ($row['ok'] ?? 0); + if ($total <= 0) { + return 0.0; + } + return round(100.0 * $ok / $total, 1); + } + + private static function avgBitratePerDay( + int $startMs, + ?int $companyId, + ?int $userId, + string $column + ): array { + $allowed = ['avg_kbps', 'recv_kbps']; + if (!in_array($column, $allowed, true)) { + return []; + } + return self::avgNumericPerDay($startMs, $companyId, $userId, $column); + } + + private static function crashesPerDay(int $startMs, ?int $companyId): array { + if (!Database::tableExists(Database::pdo(), 'reports')) { + return []; + } + $where = ['received_at_ms >= ?']; + $params = [$startMs]; + if ($companyId !== null && $companyId > 0) { + $where[] = 'company_id = ?'; + $params[] = $companyId; + } + return self::countByDay('reports', 'received_at_ms', $where, $params); + } + + private static function ticketsPerDay(int $startMs, ?int $companyId): array { + $where = ['opened_at_ms >= ?']; + $params = [$startMs]; + if ($companyId !== null && $companyId > 0) { + $where[] = 'company_id = ?'; + $params[] = $companyId; + } + return self::countByDay('tickets', 'opened_at_ms', $where, $params); + } + + private static function topCrashFingerprints(int $startMs, int $limit): array { + if (!Database::tableExists(Database::pdo(), 'reports')) { + return []; + } + $sql = 'SELECT fingerprint, COUNT(*) AS c, MAX(id) AS last_id + FROM reports WHERE received_at_ms >= ? + GROUP BY fingerprint ORDER BY c DESC LIMIT ' . (int) $limit; + $stmt = Database::pdo()->prepare($sql); + $stmt->execute([$startMs]); + $base = Auth::basePath(); + $out = []; + foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) { + $fp = (string) ($row['fingerprint'] ?? ''); + $id = (int) ($row['last_id'] ?? 0); + $out[] = [ + 'label' => $fp, + 'value' => (int) ($row['c'] ?? 0), + 'href' => $id > 0 ? $base . '/?view=report&id=' . $id : null, + ]; + } + return $out; + } + + private static function crashesByAppVersion(int $startMs): array { + if (!Database::tableExists(Database::pdo(), 'reports')) { + return []; + } + $sql = "SELECT COALESCE(NULLIF(app_version, ''), 'unknown') AS k, COUNT(*) AS c + FROM reports WHERE received_at_ms >= ? + GROUP BY k ORDER BY c DESC LIMIT 12"; + $stmt = Database::pdo()->prepare($sql); + $stmt->execute([$startMs]); + $out = []; + foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) { + $out[] = ['label' => (string) ($row['k'] ?? 'unknown'), 'value' => (int) ($row['c'] ?? 0)]; + } + return $out; + } + + private static function avgNtpCorrection(int $startMs, ?int $companyId, ?int $userId): ?float { + $where = ['started_at_ms >= ?', 'ntp_correction_s IS NOT NULL']; + $params = [$startMs]; + self::appendScope($where, $params, $companyId, $userId); + $sql = 'SELECT AVG(ABS(ntp_correction_s)) AS avg_s FROM graph_sessions WHERE ' + . implode(' AND ', $where); + $stmt = Database::pdo()->prepare($sql); + $stmt->execute($params); + $v = $stmt->fetchColumn(); + return $v === false || $v === null ? null : round((float) $v, 2); + } + + private static function breakdown(string $column, int $startMs, ?int $companyId, ?int $userId): array { + $allowed = ['ntp_source', 'install_source', 'transport', 'app_version']; + if (!in_array($column, $allowed, true)) { + return []; + } + $where = ['started_at_ms >= ?']; + $params = [$startMs]; + if ($column === 'app_version') { + $where[] = "app_version IS NOT NULL"; + $where[] = "app_version <> ''"; + } + self::appendScope($where, $params, $companyId, $userId); + $sql = 'SELECT ' . $column . ' AS k, COUNT(*) AS c FROM graph_sessions WHERE ' + . implode(' AND ', $where) . ' GROUP BY ' . $column . ' ORDER BY c DESC'; + $stmt = Database::pdo()->prepare($sql); + $stmt->execute($params); + $out = []; + foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) { + $label = (string) ($row['k'] ?? 'unknown'); + if ($label === '') { + $label = 'unknown'; + } + $out[] = ['label' => $label, 'value' => (int) ($row['c'] ?? 0)]; + } + return $out; + } + + private static function msDayExpr(string $msColumn): string { + if (Database::isMysql()) { + return 'DATE(FROM_UNIXTIME(FLOOR(' . $msColumn . ' / 1000)))'; + } + return "strftime('%Y-%m-%d', " . $msColumn . " / 1000, 'unixepoch')"; + } + + private static function countByDay(string $table, string $msColumn, array $where, array $params): array { + $day = self::msDayExpr($msColumn); + $sql = 'SELECT ' . $day . ' AS d, COUNT(*) AS c FROM ' . $table . ' WHERE ' + . implode(' AND ', $where) . ' GROUP BY d ORDER BY d'; + $stmt = Database::pdo()->prepare($sql); + $stmt->execute($params); + $bucket = []; + foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) { + $bucket[(string) ($row['d'] ?? '')] = (int) ($row['c'] ?? 0); + } + return self::assocToSeries($bucket); + } + + private static function avgNumericPerDay( + int $startMs, + ?int $companyId, + ?int $userId, + string $column + ): array { + $where = ['started_at_ms >= ?']; + $params = [$startMs]; + self::appendScope($where, $params, $companyId, $userId); + $day = self::msDayExpr('started_at_ms'); + $sql = 'SELECT ' . $day . ' AS d, AVG(' . $column . ') AS v FROM graph_sessions WHERE ' + . implode(' AND ', $where) . ' GROUP BY d ORDER BY d'; + $stmt = Database::pdo()->prepare($sql); + $stmt->execute($params); + $avg = []; + foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) { + $avg[(string) ($row['d'] ?? '')] = round((float) ($row['v'] ?? 0), 1); + } + return self::assocToSeries($avg); + } + + private static function appendScope(array &$where, array &$params, ?int $companyId, ?int $userId): void { + if ($companyId !== null && $companyId > 0) { + $where[] = 'company_id = ?'; + $params[] = $companyId; + } + if ($userId !== null && $userId > 0) { + $where[] = 'user_id = ?'; + $params[] = $userId; + } + } + + private static function assocToSeries(array $assoc): array { + ksort($assoc); + $out = []; + foreach ($assoc as $k => $v) { + $out[] = ['x' => $k, 'y' => $v]; + } + return $out; + } +}