<?php
require_once __DIR__ . '/../auth.php';
if (function_exists('require_login')) require_login('login.php');
require_once __DIR__ . '/../db_bootstrap.php';
require_once __DIR__ . '/../quota_lib.php';
require_once __DIR__ . '/../mikrotik_lib.php';
require_once __DIR__ . "/../ppp_active_cache_lib.php";
require_once __DIR__ . '/../subscriber_status_lib.php';
require_once __DIR__ . '/../ua_unified_stats_lib.php';
require_once __DIR__ . '/../stats_snapshot_lib.php';
require_once __DIR__ . '/../subscriber_reality_lib.php';
require_once __DIR__ . '/../receiver_subscriber_match_lib.php';
if (session_status() === PHP_SESSION_ACTIVE) { @session_write_close(); }
header('Content-Type: application/json; charset=utf-8');
if (function_exists('ensure_db')) ensure_db(true);
$conn = $GLOBALS['conn'] ?? null;
if (!($conn instanceof mysqli)) { echo json_encode(['ok'=>false,'message'=>'db']); exit; }

// مهم للأداء: هذه الواجهة تُستدعى كل 10 ثواني من صفحة عرض المشتركين.
// لذلك لا ننفذ هنا أي فحص MikroTik عميق أو ARP/Neighbor.
// المراقبة الحقيقية تعمل من cron/mikrotik_watch_worker.php فقط.

mysqli_report(MYSQLI_REPORT_OFF);

function col_exists_live(mysqli $conn, string $table, string $col): bool {
  static $cache = [];
  $k = $table . '|' . $col;
  if (isset($cache[$k])) return $cache[$k];
  $t = $conn->real_escape_string($table);
  $c = $conn->real_escape_string($col);
  $rs = @$conn->query("SHOW COLUMNS FROM `{$t}` LIKE '{$c}'");
  return $cache[$k] = (bool)($rs && $rs->num_rows > 0);
}
function time_ago_live($datetime) {
  if (!$datetime) return "—";
  $ts = strtotime($datetime);
  if (!$ts || $ts < 946684800) return "—";
  $diff = time() - $ts;
  if ($diff < 60) return "الآن";
  if ($diff < 3600) return "منذ " . floor($diff/60) . " دقيقة";
  if ($diff < 86400) return "منذ " . floor($diff/3600) . " ساعة";
  if ($diff < 604800) return "منذ " . floor($diff/86400) . " يوم";
  return date("Y-m-d H:i", $ts);
}



function live_sync_gate(string $name, int $ttl = 25): bool {
  $base = __DIR__ . '/../cache/runtime';
  if (!is_dir($base)) { @mkdir($base, 0777, true); }
  $safe = preg_replace('/[^a-z0-9_\-]+/i', '_', $name);
  $path = rtrim($base, DIRECTORY_SEPARATOR) . DIRECTORY_SEPARATOR . $safe . '.lock';
  $now = time();
  $last = @filemtime($path);
  if ($last && (($now - (int)$last) < $ttl)) return false;
  @touch($path);
  return true;
}

function live_status_cache_dir(): string {
  $dir = __DIR__ . '/../cache/runtime/subscribers-live';
  if (!is_dir($dir)) { @mkdir($dir, 0770, true); }
  return $dir;
}

function live_status_cache_read(string $path): ?array {
  if (!is_file($path)) return null;
  $raw = @file_get_contents($path);
  if (!is_string($raw) || $raw === '') return null;
  $payload = json_decode($raw, true);
  return is_array($payload) && !empty($payload['ok']) && isset($payload['revision']) ? $payload : null;
}

function live_status_cache_write(string $path, array $payload): void {
  $json = json_encode($payload, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
  if (!is_string($json) || $json === '') return;
  $tmp = $path . '.' . getmypid() . '.' . bin2hex(random_bytes(3)) . '.tmp';
  if (@file_put_contents($tmp, $json, LOCK_EX) === false) return;
  if (!@rename($tmp, $path)) @unlink($tmp);
}

function live_status_send(array $payload, string $since = '', bool $cacheHit = false): void {
  $revision = (string)($payload['revision'] ?? '');
  header('X-Subscribers-Revision: ' . $revision);
  header('X-Subscribers-Cache: ' . ($cacheHit ? 'HIT' : 'MISS'));
  if ($since !== '' && $revision !== '' && hash_equals($revision, $since)) {
    echo json_encode([
      'ok' => true,
      'unchanged' => true,
      'revision' => $revision,
      'generated_at' => (string)($payload['generated_at'] ?? ''),
      'poll_mode' => 'cached_snapshot',
    ], JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
    exit;
  }
  $payload['cache_hit'] = $cacheHit;
  echo json_encode($payload, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
  exit;
}

function usage_batch_live(mysqli $conn, array $subs, string $cycleMonth): array {
  $map = [];
  if (!$subs) return $map;

  $byId = [];
  $byUser = [];
  foreach ($subs as $sub) {
    $sid = (int)($sub['id'] ?? 0);
    $u = trim((string)($sub['ppp_username'] ?? ''));
    if ($u === '') $u = trim((string)($sub['username'] ?? ''));
    $u = preg_replace('/@.*/', '', $u);
    if ($sid > 0) $byId[$sid] = true;
    if ($u !== '') $byUser[$u] = true;
  }

  if (col_exists_live($conn, 'ppp_usage_monthly', 'subscriber_id') && col_exists_live($conn, 'ppp_usage_monthly', 'month_key') && $byId) {
    $ids = implode(',', array_map('intval', array_keys($byId)));
    $m = $conn->real_escape_string($cycleMonth);
    $rs = @$conn->query("SELECT subscriber_id, SUM(used_bytes) used_bytes FROM ppp_usage_monthly WHERE month_key='{$m}' AND subscriber_id IN ({$ids}) GROUP BY subscriber_id");
    if ($rs) while ($r = $rs->fetch_assoc()) $map['id:' . (int)$r['subscriber_id']] = (int)($r['used_bytes'] ?? 0);
  }

  if (col_exists_live($conn, 'ppp_usage_monthly', 'username') && col_exists_live($conn, 'ppp_usage_monthly', 'cycle_month') && $byUser) {
    $in = "'" . implode("','", array_map([$conn, 'real_escape_string'], array_keys($byUser))) . "'";
    $m = $conn->real_escape_string($cycleMonth);
    $rs = @$conn->query("SELECT username, SUM(used_bytes) used_bytes FROM ppp_usage_monthly WHERE cycle_month='{$m}' AND username IN ({$in}) GROUP BY username");
    if ($rs) while ($r = $rs->fetch_assoc()) $map['user:' . trim((string)$r['username'])] = (int)($r['used_bytes'] ?? 0);
  }

  return $map;
}

// Keep this endpoint read-only and fast; usage sync belongs to the background worker.

$filter = trim((string)($_GET['filter'] ?? 'all'));
$q = trim((string)($_GET['q'] ?? ''));
$forceRefresh = ((string)($_GET['force'] ?? '') === '1');
$sinceRevision = preg_replace('/[^a-f0-9]/i', '', (string)($_GET['since'] ?? ''));
$cacheTtl = 12;
$cacheKey = hash('sha256', $filter . "\n" . $q);
$cacheDir = live_status_cache_dir();
$cachePath = $cacheDir . DIRECTORY_SEPARATOR . $cacheKey . '.json';
$lockPath = $cacheDir . DIRECTORY_SEPARATOR . $cacheKey . '.lock';
$cachedPayload = live_status_cache_read($cachePath);
$cacheAge = is_file($cachePath) ? max(0, time() - (int)@filemtime($cachePath)) : PHP_INT_MAX;
if (!$forceRefresh && $cachedPayload !== null && $cacheAge <= $cacheTtl) {
  live_status_send($cachedPayload, $sinceRevision, true);
}

$cacheLock = @fopen($lockPath, 'c+');
$hasCacheLock = is_resource($cacheLock) && @flock($cacheLock, LOCK_EX | LOCK_NB);
if (!$hasCacheLock && $cachedPayload !== null) {
  // Another request is already rebuilding the snapshot. Serve the last complete
  // atomic snapshot instead of making every open tab repeat the expensive work.
  live_status_send($cachedPayload, $sinceRevision, true);
}

// UI polling must never open a MikroTik connection. The background worker owns
// the source caches; this endpoint only composes their last atomic snapshots.
$subscriberReality = ua_subscriber_reality_snapshot($conn, 3600);
$pppUnifiedCache = (array)($subscriberReality['raw'] ?? []);
$pppActiveSet = (array)($subscriberReality['active'] ?? []);
$hasPppObservation = !empty($subscriberReality['has_observation']);
$pppActiveRowMap = (array)($subscriberReality['active_rows'] ?? []);
$rxMatchSnapshot = function_exists('ua_receiver_match_snapshot') ? ua_receiver_match_snapshot($conn, 300, false) : [];
$rxMatchBySubscriber = function_exists('ua_receiver_match_by_subscriber') ? ua_receiver_match_by_subscriber($rxMatchSnapshot) : [];

$where = [];
if (col_exists_live($conn,'subscribers','is_deleted')) $where[] = 'COALESCE(is_deleted,0)=0';
$disabledConds = [];
if (col_exists_live($conn,'subscribers','ppp_is_disabled')) $disabledConds[] = "COALESCE(ppp_is_disabled,0)=1";
if (col_exists_live($conn,'subscribers','mt_disabled')) $disabledConds[] = "COALESCE(mt_disabled,0)=1";
if (col_exists_live($conn,'subscribers','is_cut')) $disabledConds[] = "COALESCE(is_cut,0)=1";
if (col_exists_live($conn,'subscribers','speed_state')) $disabledConds[] = "UPPER(COALESCE(speed_state,''))='CUT'";
if (col_exists_live($conn,'subscribers','status')) $disabledConds[] = "UPPER(COALESCE(status,'')) IN ('CUT','DISABLED','STOPPED')";
$limitedConds = [];
if (col_exists_live($conn,'subscribers','speed_state')) $limitedConds[] = "UPPER(COALESCE(speed_state,'')) IN ('LIMIT','LIMITED','REDUCED','THROTTLED')";
if (col_exists_live($conn,'subscribers','current_speed') && col_exists_live($conn,'subscribers','base_speed')) {
  $limitedConds[] = "(COALESCE(current_speed,'')<>'' AND COALESCE(base_speed,'')<>'' AND COALESCE(current_speed,'')<>COALESCE(base_speed,''))";
}
if ($filter === 'limited') $where[] = $limitedConds ? '(' . implode(' OR ', $limitedConds) . ')' : '0=1';
if ($q !== '') {
  $qe = $conn->real_escape_string($q);
  $like = "%{$qe}%";
  $conds = [];
  if (ctype_digit($q)) $conds[] = 'id=' . (int)$q;
  $conds[] = "username LIKE '{$like}'";
  foreach (['ppp_username','full_name','phone'] as $col) if (col_exists_live($conn,'subscribers',$col)) $conds[] = "{$col} LIKE '{$like}'";
  $where[] = '(' . implode(' OR ', $conds) . ')';
}
$whereSql = $where ? ('WHERE ' . implode(' AND ', $where)) : '';
$hasPppUser = col_exists_live($conn,'subscribers','ppp_username');
$userCols = $hasPppUser ? ', ppp_username, username' : ', username';
$rxCols = '';
foreach (['airmax_device_name','airmax_matched_mac'] as $rxCol) {
  $rxCols .= col_exists_live($conn,'subscribers',$rxCol) ? ", {$rxCol}" : ", NULL AS {$rxCol}";
}
$statusCols = '';
foreach (['speed_state','current_speed','base_speed','status'] as $statusCol) {
  $statusCols .= col_exists_live($conn,'subscribers',$statusCol) ? ", {$statusCol}" : ", '' AS {$statusCol}";
}
$sql = "SELECT id, COALESCE(is_online,0) AS is_online, CASE WHEN COALESCE(ppp_is_disabled,0)=1 OR COALESCE(mt_disabled,0)=1 OR UPPER(COALESCE(speed_state,''))='CUT' THEN 1 ELSE 0 END AS is_disabled" . (col_exists_live($conn,'subscribers','last_seen_at') ? ", last_seen_at" : ", NULL AS last_seen_at") . (col_exists_live($conn,'subscribers','last_disconnect_at') ? ", last_disconnect_at" : ", NULL AS last_disconnect_at") . (col_exists_live($conn,'subscribers','ppp_last_cut_at') ? ", ppp_last_cut_at" : ", NULL AS ppp_last_cut_at") . "{$userCols}{$statusCols}{$rxCols} FROM subscribers {$whereSql} ORDER BY id";
$rows = [];
$rawSubs = [];
$rs = @$conn->query($sql);
if ($rs) while ($r = $rs->fetch_assoc()) {
  $sid = (int)$r['id'];
  $rawSubs[$sid] = $r;
}
$usageMap = usage_batch_live($conn, array_values($rawSubs), date('Y-m'));
$mtRealityMap = (array)($subscriberReality['secrets'] ?? []);
foreach ($rawSubs as $sid => $r) {
  $uJoin = trim((string)($r['ppp_username'] ?? ''));
  if ($uJoin === '') $uJoin = trim((string)($r['username'] ?? ''));
  $uJoin = preg_replace('/@.*/', '', $uJoin);
  $runtimeState = ua_subscriber_reality_state($r, $subscriberReality);
  $cacheIsOnline = (int)$runtimeState['is_online'];
  $usedBytes = (int)($usageMap['id:' . $sid] ?? ($usageMap['user:' . $uJoin] ?? 0));
  $isDisabled = (int)$runtimeState['is_disabled'];
  if ($filter === 'online' && $cacheIsOnline !== 1) continue;
  if ($filter === 'offline' && $cacheIsOnline === 1) continue;
  if ($filter === 'disabled' && $isDisabled !== 1) continue;
  $activeRow = ($uJoin !== '' && isset($pppActiveRowMap[$uJoin]) && is_array($pppActiveRowMap[$uJoin])) ? $pppActiveRowMap[$uJoin] : null;
  $statusInfo = sub_status_info($cacheIsOnline === 1, $isDisabled === 1, $r, $activeRow);

  $mtReality = ($uJoin !== '' && isset($mtRealityMap[$uJoin]) && is_array($mtRealityMap[$uJoin])) ? $mtRealityMap[$uJoin] : [];
  $mtLastLogout = trim((string)($mtReality['last_logged_out'] ?? ''));
  $mtLastLogoutTs = $mtLastLogout !== '' ? strtotime($mtLastLogout) : false;

  $lastSeenRaw = (string)($r['last_seen_at'] ?? '');
  if ($cacheIsOnline !== 1 && $mtLastLogoutTs && $mtLastLogoutTs >= 946684800) {
    $localLsTs = $lastSeenRaw !== '' ? strtotime($lastSeenRaw) : false;
    if (!$localLsTs || $mtLastLogoutTs >= $localLsTs) $lastSeenRaw = date('Y-m-d H:i:s', $mtLastLogoutTs);
  }
  $lastSeenTs = $lastSeenRaw !== '' ? strtotime($lastSeenRaw) : false;
  $lastSeenFull = ($lastSeenTs && $lastSeenTs >= 946684800) ? date('Y-m-d H:i:s', $lastSeenTs) : '';

  $lastCutRaw = (string)($r['ppp_last_cut_at'] ?? '');
  if ($lastCutRaw === '') $lastCutRaw = (string)($r['last_disconnect_at'] ?? '');
  if ($mtLastLogoutTs && $mtLastLogoutTs >= 946684800) {
    $localCutTs = $lastCutRaw !== '' ? strtotime($lastCutRaw) : false;
    if (!$localCutTs || $mtLastLogoutTs >= $localCutTs) $lastCutRaw = date('Y-m-d H:i:s', $mtLastLogoutTs);
  }
  $lastCutTs = $lastCutRaw !== '' ? strtotime($lastCutRaw) : false;
  $lastCutDisplay = ($lastCutTs && $lastCutTs >= 946684800) ? date('Y-m-d H:i', $lastCutTs) : '—';

  $rxAuto = $rxMatchBySubscriber[$sid] ?? [];
  $receiverName = trim((string)($r['airmax_device_name'] ?? ''));
  if ($receiverName === '' && is_array($rxAuto)) $receiverName = trim((string)($rxAuto['device_name'] ?? ''));
  $receiverMac = trim((string)($r['airmax_matched_mac'] ?? ''));
  if ($receiverMac === '' && is_array($rxAuto)) $receiverMac = trim((string)($rxAuto['mac'] ?? ''));

  $rows[$sid] = [
    'is_online'        => $cacheIsOnline,
    'is_disabled'      => $isDisabled,
    'status_key'       => (string)($statusInfo['key'] ?? ''),
    'status_class'     => (string)($statusInfo['class'] ?? ''),
    'status_label'     => (string)($statusInfo['label'] ?? ''),
    'used_bytes'       => $usedBytes,
    'used_label'       => 'GB ' . number_format($usedBytes / 1073741824, 2),
    'last_seen_label'  => ($cacheIsOnline === 1) ? 'الآن' : time_ago_live($lastSeenFull),
    'last_seen_full'   => $lastSeenFull,
    'last_cut_display' => $lastCutDisplay,
    'receiver_name'    => $receiverName !== '' ? $receiverName : '—',
    'receiver_auto'    => $receiverName !== '' ? 1 : 0,
    'receiver_title'   => $receiverName !== '' ? ('تمت المطابقة تلقائياً عبر MAC: ' . $receiverMac) : 'لا توجد مطابقة بعد',
  ];
}
$liveOnlineCount = 0;
foreach ($rows as $_sid => $_row) { if ((int)($_row['is_online'] ?? 0) === 1) $liveOnlineCount++; }

$stats = function_exists('ua_stats_snapshot_read') ? ua_stats_snapshot_read(86400) : null;
$hadStatsSnapshot = is_array($stats);
if (!is_array($stats)) {
  $total = 0;
  $disabled = 0;
  $countSql = "SELECT COUNT(*) AS total";
  if ($disabledConds) $countSql .= ", SUM(CASE WHEN " . implode(' OR ', $disabledConds) . " THEN 1 ELSE 0 END) AS disabled";
  else $countSql .= ", 0 AS disabled";
  $countSql .= " FROM subscribers";
  if (col_exists_live($conn,'subscribers','is_deleted')) $countSql .= " WHERE COALESCE(is_deleted,0)=0";
  $countRs = @$conn->query($countSql);
  if ($countRs && ($countRow = $countRs->fetch_assoc())) {
    $total = (int)($countRow['total'] ?? 0);
    $disabled = (int)($countRow['disabled'] ?? 0);
  }
  $onlineFromCache = (int)($pppUnifiedCache['online'] ?? count($pppActiveSet));
  $stats = [
    'total' => $total,
    'online' => $onlineFromCache,
    'offline' => max(0, $total - $onlineFromCache),
    'disabled' => $disabled,
    'financial_cut' => $disabled,
    'unsafe_speed' => 0,
    'source' => 'database_and_ppp_cache',
    'snapshot_source' => 'database_and_ppp_cache',
  ];
}
$isWholeSubscriberSet = ($filter === 'all' && $q === '');
if ($isWholeSubscriberSet && !$hadStatsSnapshot) {
  $stats['online'] = $liveOnlineCount;
  if (isset($stats['total'])) $stats['offline'] = max(0, (int)$stats['total'] - $liveOnlineCount);
}
$stats['online_source'] = (string)($pppUnifiedCache['source'] ?? ($stats['online_source'] ?? 'ppp_active_cache'));
$stats['source'] = (string)($stats['online_source'] ?? ($pppUnifiedCache['source'] ?? 'ppp_active_cache_read_only'));
$stats['age'] = (int)($pppUnifiedCache['data_age'] ?? $pppUnifiedCache['age'] ?? 0);
$stats['updated_at'] = date('Y-m-d H:i:s');
$stats['poll_mode'] = 'cached_snapshot';
$stats['router_online'] = array_key_exists('router_online', $pppUnifiedCache) ? $pppUnifiedCache['router_online'] : null;
$stats['live_data_reliable'] = !empty($pppUnifiedCache['live_data_reliable']);
$stats['data_available'] = $hasPppObservation;
$stats['connection_state'] = (string)($pppUnifiedCache['connection_state'] ?? ($stats['live_data_reliable'] ? 'online' : 'unknown'));
$revisionStats = $stats;
unset($revisionStats['age'], $revisionStats['updated_at']);
$revisionJson = json_encode([$rows, $revisionStats], JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
$payload = [
  'ok' => true,
  'rows' => $rows,
  'stats' => $stats,
  'revision' => sha1(is_string($revisionJson) ? $revisionJson : ''),
  'generated_at' => date('Y-m-d H:i:s'),
  'poll_mode' => 'cached_snapshot',
];
live_status_cache_write($cachePath, $payload);
if ($hasCacheLock && is_resource($cacheLock)) @flock($cacheLock, LOCK_UN);
if (is_resource($cacheLock)) @fclose($cacheLock);
live_status_send($payload, $sinceRevision, false);
