<?php
// get_markets_stream.php (PDO) — SSE Live Stream
// ✅ Live SSE
// ✅ Allow-list من DB (asset_pairs + assets.code + assets.icon_url) — بدون قوائم ثابتة
// ✅ Binance server-side + Failover
// ✅ Chunking للرموز لتفادي طول URL
// ✅ File-cache قصير لتقليل الضغط مع تعدد الاتصالات
// ✅ إغلاق الاتصال بعد تكرار فشل المصدر لتحفيز fallback في الواجهة
// ✅ PHP 7.4+

if (session_status() === PHP_SESSION_NONE) session_start();

/* ================= CORS ================= */
$allowedOrigins = [
  'https://eazzybit.com',
  'https://www.eazzybit.com',
];
$reqOrigin = $_SERVER['HTTP_ORIGIN'] ?? '';
if ($reqOrigin && in_array($reqOrigin, $allowedOrigins, true)) {
  header("Access-Control-Allow-Origin: {$reqOrigin}");
  header("Vary: Origin");
  header("Access-Control-Allow-Credentials: true");
}
header("Access-Control-Allow-Headers: Content-Type, Authorization, X-Requested-With");
header("Access-Control-Allow-Methods: GET, OPTIONS");
header("Access-Control-Max-Age: 600");

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

/* ================= SSE headers ================= */
header("Content-Type: text/event-stream; charset=UTF-8");
header("Cache-Control: no-cache, no-store, must-revalidate");
header("Pragma: no-cache");
header("Connection: keep-alive");
header("X-Accel-Buffering: no"); // لمنع buffering في Nginx إن وُجد

/* ================= DEBUG ================= */
$DEBUG_MODE = false;
if ($DEBUG_MODE) { ini_set('display_errors', 1); error_reporting(E_ALL); }
else { ini_set('display_errors', 0); error_reporting(0); }

/* ================= Helpers ================= */
function sse_send(string $event, array $data): void {
  echo "event: {$event}\n";
  echo "data: " . json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) . "\n\n";
  @ob_flush();
  @flush();
}

function safe_upper($v): string { return strtoupper(trim((string)$v)); }

function read_input(): array { return $_GET ?? []; }

function log_error(Throwable $e): void {
  $dir = __DIR__ . '/../../logs';
  if (!is_dir($dir)) @mkdir($dir, 0755, true);
  $file = $dir . '/api_errors_' . date('Y-m-d') . '.log';
  $line = "[" . date('Y-m-d H:i:s') . "] " . $e->getMessage() . " | " . $e->getFile() . ":" . $e->getLine() . PHP_EOL;
  @file_put_contents($file, $line, FILE_APPEND | LOCK_EX);
}

function ends_with(string $haystack, string $needle): bool {
  $len = strlen($needle);
  if ($len === 0) return true;
  return substr($haystack, -$len) === $needle;
}

/* ================= DB ================= */
require_once __DIR__ . '/../config.php';
if (!isset($conn) || !($conn instanceof PDO)) {
  sse_send('markets', ['success'=>false, 'message'=>'اتصال قاعدة البيانات غير متوفر.']);
  exit;
}

/* ================= AUTH ================= */
function resolveUserIdByToken(PDO $conn, array $in): int {
  $sid = isset($_SESSION['user_id']) ? (int)$_SESSION['user_id'] : 0;
  if ($sid > 0) return $sid;

  $tok = trim((string)($in['user_token'] ?? ''));
  if ($tok === '') return 0;

  $st = $conn->prepare("
    SELECT user_id
    FROM user_sessions
    WHERE user_token = :tok
      AND is_online = 1
    LIMIT 1
  ");
  $st->execute([':tok' => $tok]);
  $row = $st->fetch(PDO::FETCH_ASSOC);
  return $row ? (int)$row['user_id'] : 0;
}

/* ================= Cache (file) ================= */
function cache_dir(): string {
  $dir = __DIR__ . '/../../cache';
  if (!is_dir($dir)) @mkdir($dir, 0755, true);
  return $dir;
}
function cache_file(string $key): string { return cache_dir() . '/' . $key . '.json'; }

function cache_get(string $key, int $ttlSec): ?array {
  if ($ttlSec <= 0) return null;
  $file = cache_file($key);
  if (!is_file($file)) return null;
  $mt = @filemtime($file);
  if (!$mt || (time() - $mt) > $ttlSec) return null;
  $raw = @file_get_contents($file);
  if (!$raw) return null;
  $json = json_decode($raw, true);
  return is_array($json) ? $json : null;
}

function cache_set(string $key, array $value): void {
  $file = cache_file($key);
  @file_put_contents($file, json_encode($value, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES), LOCK_EX);
}

/* ================= Allowed pairs from DB =================
   asset_pairs: base_asset_id, quote_asset_id, tv_symbol
   assets: code, icon_url, is_active
=========================================================== */
function allowed_pairs_from_db(PDO $conn, string $quote, int $limit = 200): array {
  $quote = safe_upper($quote ?: 'USDT');
  $limit = (int)$limit;
  if ($limit < 1) $limit = 1;
  if ($limit > 500) $limit = 500;

  $sql = "
    SELECT
      ap.tv_symbol,
      b.code AS base_code,
      q.code AS quote_code,
      b.icon_url AS base_icon_url
    FROM asset_pairs ap
    JOIN assets b ON b.id = ap.base_asset_id AND b.is_active = 1
    JOIN assets q ON q.id = ap.quote_asset_id AND q.is_active = 1
    WHERE UPPER(q.code) = :quote
    ORDER BY ap.base_asset_id ASC
    LIMIT {$limit}
  ";

  $st = $conn->prepare($sql);
  $st->execute([':quote' => $quote]);

  $symbols = [];
  $meta = []; // symbol => [base, quote, icon_url]

  while ($row = $st->fetch(PDO::FETCH_ASSOC)) {
    $tv = safe_upper($row['tv_symbol'] ?? '');
    $bc = safe_upper($row['base_code'] ?? '');
    $qc = safe_upper($row['quote_code'] ?? '');
    $icon = trim((string)($row['base_icon_url'] ?? ''));

    $sym = $tv !== '' ? $tv : (($bc !== '' && $qc !== '') ? ($bc . $qc) : '');
    $sym = preg_replace('/[^A-Z0-9_]/', '', $sym);

    if ($sym === '') continue;
    if (!ends_with($sym, $quote)) continue;

    $symbols[] = $sym;
    $meta[$sym] = [
      'base' => $bc !== '' ? $bc : $sym,
      'quote' => $qc !== '' ? $qc : $quote,
      'icon_url' => ($icon !== '' ? $icon : null),
    ];
  }

  $symbols = array_values(array_unique($symbols));
  if (count($symbols) > 200) $symbols = array_slice($symbols, 0, 200);

  // تنظيف meta بما يتوافق مع الرموز النهائية
  $meta2 = [];
  foreach ($symbols as $s) { $meta2[$s] = $meta[$s] ?? ['base'=>null,'quote'=>$quote,'icon_url'=>null]; }

  return ['symbols' => $symbols, 'meta' => $meta2];
}

/* ================= Binance fetch (server-side) ================= */
function http_get_json(string $url, int $timeoutSec = 8): array {
  $ch = curl_init($url);
  curl_setopt_array($ch, [
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_FOLLOWLOCATION => true,
    CURLOPT_CONNECTTIMEOUT => $timeoutSec,
    CURLOPT_TIMEOUT => $timeoutSec,
    CURLOPT_SSL_VERIFYPEER => true,
    CURLOPT_HTTPHEADER => [
      'Accept: application/json',
      'User-Agent: EazzyBit-MarketStream/1.0'
    ],
  ]);
  $body = curl_exec($ch);
  $err  = curl_error($ch);
  $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
  curl_close($ch);

  if ($body === false) throw new RuntimeException("CURL_ERROR: {$err}");
  $json = json_decode($body, true);
  if (!is_array($json)) throw new RuntimeException("BAD_JSON_FROM_SOURCE");
  if ($code < 200 || $code >= 300) {
    $msg = isset($json['msg']) ? (string)$json['msg'] : "HTTP_{$code}";
    throw new RuntimeException("SOURCE_ERROR: {$msg}");
  }
  return $json;
}

function fetch_binance_24hr(array $symbols, int $timeout = 10): array {
  $symbolsJson = json_encode($symbols, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
  $path = '/api/v3/ticker/24hr?symbols=' . urlencode($symbolsJson);

  $hosts = [
    'https://api.binance.com',
    'https://api1.binance.com',
    'https://api2.binance.com',
    'https://api3.binance.com',
    'https://data-api.binance.vision',
  ];

  $lastErr = null;
  foreach ($hosts as $h) {
    try {
      return http_get_json($h . $path, $timeout);
    } catch (Throwable $e) {
      $lastErr = $e;
    }
  }
  throw new RuntimeException("MARKET_SOURCE_UNAVAILABLE" . ($lastErr ? (": " . $lastErr->getMessage()) : ""));
}

function fetch_binance_24hr_multi(array $symbols, int $timeout = 10, int $chunkSize = 80): array {
  $symbols = array_values($symbols);
  if (!$symbols) return [];
  if ($chunkSize < 1) $chunkSize = 80;

  $out = [];
  foreach (array_chunk($symbols, $chunkSize) as $chunk) {
    $part = fetch_binance_24hr($chunk, $timeout);
    if (is_array($part)) {
      foreach ($part as $row) $out[] = $row;
    }
  }
  return $out;
}

/* ================= Main ================= */
try {
  @set_time_limit(0);
  @ignore_user_abort(true);

  $in = read_input();

  $userId = resolveUserIdByToken($conn, $in);
  if ($userId <= 0) {
    http_response_code(401);
    sse_send('markets', ['success'=>false, 'unauthorized'=>true, 'message'=>'غير مصرح. الرجاء تسجيل الدخول.']);
    exit;
  }

  $quote = safe_upper($in['quote'] ?? 'USDT');
  if ($quote === '') $quote = 'USDT';

  // interval بالمللي ثانية
  $interval = (int)($in['interval'] ?? 1500);
  if ($interval < 1000) $interval = 1000;
  if ($interval > 10000) $interval = 10000;

  // Allow-list من DB
  $allow = allowed_pairs_from_db($conn, $quote, 200);
  $ALLOWED = $allow['symbols'];
  $META = $allow['meta'];

  if (!$ALLOWED) {
    sse_send('markets', [
      'success' => true,
      'message' => 'لا توجد أزواج سوق مفعّلة في قاعدة البيانات لهذا الـ quote.',
      'items' => [],
      'count' => 0,
      'user_id' => $userId,
      'ts' => time(),
      'note' => 'empty_allowlist'
    ]);
    exit;
  }

  // (اختياري) إذا أرسل العميل symbols، نفلترها ضمن المسموح
  $symbolsRaw = $in['symbols'] ?? null;
  $symbols = [];
  if (is_array($symbolsRaw)) {
    $symbols = $symbolsRaw;
  } else {
    $try = json_decode((string)$symbolsRaw, true);
    if (is_array($try)) $symbols = $try;
    else {
      $csv = trim((string)$symbolsRaw);
      if ($csv !== '') $symbols = array_map('trim', explode(',', $csv));
    }
  }

  $clean = [];
  foreach ($symbols as $s) {
    $sym = preg_replace('/[^A-Z0-9_]/', '', safe_upper($s));
    if ($sym !== '') $clean[] = $sym;
  }
  $clean = array_values(array_unique($clean));
  if (count($clean) > 200) $clean = array_slice($clean, 0, 200);

  $final = $clean
    ? array_values(array_filter($clean, fn($x)=>in_array($x,$ALLOWED,true)))
    : $ALLOWED;

  if (!$final) {
    sse_send('markets', ['success'=>true, 'message'=>'لا توجد رموز مسموحة بعد الفلترة.', 'items'=>[], 'count'=>0, 'ts'=>time()]);
    exit;
  }

  // ping أولي
  sse_send('ping', ['ts'=>time()]);

  // كاش stream مشترك: TTL ثواني = ceil(interval/1000) (حد أقصى 5)
  $ttlSec = (int)ceil($interval / 1000);
  if ($ttlSec < 1) $ttlSec = 1;
  if ($ttlSec > 5) $ttlSec = 5;

  $cacheKey = 'markets_stream_v1_' . md5($quote . '|' . implode(',', $final));

  $lastHash = '';
  $tick = 0;
  $srcFail = 0;

  while (!connection_aborted()) {
    $tick++;

    try {
      $cached = cache_get($cacheKey, $ttlSec);
      if (is_array($cached) && isset($cached['items']) && is_array($cached['items'])) {
        $items = $cached['items'];
      } else {
        $data = fetch_binance_24hr_multi($final, 10, 80);

        $items = [];
        foreach ($data as $x) {
          $sym = safe_upper($x['symbol'] ?? '');
          if ($sym === '') continue;

          $base = $sym;
          $q = $quote;
          $iconUrl = null;

          if (isset($META[$sym])) {
            $base = (string)($META[$sym]['base'] ?? $base);
            $q = (string)($META[$sym]['quote'] ?? $q);
            $iconUrl = $META[$sym]['icon_url'] ?? null;
          } else {
            if ($quote && ends_with($sym, $quote)) {
              $base = substr($sym, 0, -strlen($quote));
              $q = $quote;
            } elseif (ends_with($sym, 'USDT')) {
              $base = substr($sym, 0, -4);
              $q = 'USDT';
            }
          }

          $items[] = [
            'symbol' => $sym,
            'base' => safe_upper($base),
            'quote' => safe_upper($q),
            'icon_url' => $iconUrl, // ✅ جديد
            'lastPrice' => (float)($x['lastPrice'] ?? 0),
            'priceChangePercent' => (float)($x['priceChangePercent'] ?? 0),
            'quoteVolume' => (float)($x['quoteVolume'] ?? 0),
          ];
        }

        // ترتيب ثابت لتجنّب تغيّر الـ hash بسبب اختلاف ترتيب رجوع Binance
        usort($items, fn($a,$b)=>strcmp($a['symbol'], $b['symbol']));

        cache_set($cacheKey, ['ts'=>time(), 'items'=>$items]);
      }

      $srcFail = 0;

      $payload = [
        'success' => true,
        'message' => 'live',
        'user_id' => $userId,
        'count' => count($items),
        'items' => $items,
        'ts' => time(),
      ];

      $hash = md5(json_encode($items, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES));
      if ($hash !== $lastHash || ($tick % 10) === 0) {
        $lastHash = $hash;
        sse_send('markets', $payload);
      } else {
        sse_send('ping', ['ts'=>time()]);
      }

    } catch (Throwable $srcE) {
      $srcFail += 1;
      if ($DEBUG_MODE) log_error($srcE);

      // بعد تكرار الفشل: أغلق الاتصال لتحفيز fallback في الواجهة
      if ($srcFail >= 3) {
        sse_send('markets', ['success'=>false, 'message'=>'تعذر الاتصال بمصدر السوق. سيتم التحويل للتحديث الدوري.', 'ts'=>time()]);
        exit;
      } else {
        sse_send('ping', ['ts'=>time(), 'note'=>'source_fail']);
      }
    }

    usleep($interval * 1000);
  }

} catch (Throwable $e) {
  log_error($e);
  http_response_code(500);
  $msg = 'خطأ في الخادم';
  if ($DEBUG_MODE) $msg .= ' - ' . $e->getMessage();
  sse_send('markets', ['success'=>false, 'message'=>$msg]);
  exit;
}
