tripz-llc/loction_server/driver_socket.php

<?php
/**
 * driver_socket.php
 * ==================
 * WebSocket Server للسائقين — بورت 2020
 * Internal HTTP Server  — بورت 2021
 *
 * 🚀 Level 2 Architecture (Production Ready):
 * - Event Buffering (Batching)
 * - Redis Pipelines (تقليل الـ I/O والـ Latency بشكل كبير)
 * - Memory State Cache للسائقين
 * - جميع طرق HTTP (Dispatch, Market, Force Disconnect...) موجودة بالكامل
 */

use Workerman\Worker;
use Workerman\Timer;
use Workerman\Http\Client as AsyncHttp;
use PHPSocketIO\SocketIO;
use Predis\Client as RedisClient;
use Firebase\JWT\JWT;
use Firebase\JWT\Key;

require_once __DIR__ . '/vendor/autoload.php';

// ============================================================
// ⚙️ إعدادات عامة
// ============================================================
ini_set('memory_limit', '512M');
date_default_timezone_set('Asia/Amman');

// ── Tunables (إعدادات الأداء) ──────────────────────────────────
const MIN_MOVE_METERS        = 10.0;  // GEOADD فقط إذا تحرك أكثر من 10 متر
const HMSET_SPEED_DELTA      = 1.0;   // فرق السرعة المطلوب لتحديث Redis
const HMSET_HEADING_DELTA    = 5.0;   // فرق الاتجاه المطلوب لتحديث Redis
const EXPIRE_REFRESH_SECONDS = 120;   // 2 دقيقة لتجديد الـ TTL والـ updated_at للسائق المتوقف
const FORWARD_MIN_METERS     = 15.0;  // HTTP forward للراكب
const FORWARD_MAX_SECONDS    = 3;     // أقصى مدة للـ Forward
const REDIS_BATCH_INTERVAL   = 0.5;   // تنفيذ مجمّع (Batch) كل نصف ثانية (500ms)
// ─────────────────────────────────────────────────────────────

function logMsg(string $msg): void {
    echo '[' . date('Y-m-d H:i:s') . '] ' . $msg . PHP_EOL;
}

// ============================================================
// 🎯 الإسناد بالدفعات — سبق حصري قصير ثم إطلاق للجميع
//
// ‏add_ride.php يرتّب المرشحين في Redis ثم يوقظنا. نمنح الأفضل سبقاً
// ‏حصرياً بضع ثوانٍ، وبانتهائه تُطلق الرحلة لكل السائقين المتصلين.
//
// ‏لا سُلّم تنازلي: سُلّم بين ثمانية مرشحين يعني انتظاراً قد يبلغ ثمانين
// ‏ثانية للراكب. هنا التأخير محدود بمدة السبق وحدها، مهما حدث.
//
// ‏مطفأ ما لم يضبط add_ride.php الطابور (أي ما لم يكن DISPATCH_MODE=batched).
// ============================================================
function dispatchHeadStart(): int {
    $v = (int)(getenv('DISPATCH_HEAD_START_SECONDS') ?: 5);
    return max(1, min(15, $v)); // يطابق dispatchHeadStartSeconds() في dispatch_helper.php
}

/**
 * ‏يمنح المرشح الأفضل المتصل سبقاً حصرياً، ويجدول الإطلاق للجميع بعده.
 *
 * ‏المرور على الطابور هنا ليس سُلّماً: هو اختيار أول مرشح متصل فعلاً.
 * ‏غير المتصل يُتخطّى فوراً بلا إهدار ثانية واحدة من وقت الراكب.
 */
function dispatchOfferNext($io, string $rideId): void {
    global $connectedDrivers;

    $redis = getRedis();
    if (!$redis) return;

    // ‏الرحلة قد تكون قُبلت أو أُلغيت بين تحديدين للمؤقّت.
    if (!$redis->exists("ride:$rideId:dispatch_payload")) {
        dispatchCleanup($redis, $rideId);
        return;
    }

    $headStart = dispatchHeadStart();
    $queueKey  = "ride:$rideId:dispatch_queue";

    while (($driverId = $redis->lpop($queueKey)) !== null) {
        if (!isset($connectedDrivers[$driverId])) {
            logMsg("⏭️  Ride #$rideId — تخطّي السائق #$driverId (غير متصل)");
            continue;
        }

        $payload = json_decode((string)$redis->get("ride:$rideId:dispatch_payload"), true) ?: [];

        // ‏وسم العرض: التطبيق يعرضه كـ"طلب خاص" مع عدّاد تنازلي، ليفهم
        // ‏السائق أنه صاحب الأولوية الآن لا أن الطلب وصله عشوائياً.
        $payload['offer_type']       = 'exclusive';
        $payload['offer_expires_in'] = $headStart;

        // ‏رمز الحصرية: acceptRide.php يرفض أي سائق لا يطابقه خلال السبق.
        // ‏TTL ينهي الحصرية تلقائياً حتى لو تعطّل المؤقّت.
        $redis->setex("ride:$rideId:current_offer", $headStart, $driverId);
        $redis->sadd("ride:offered_drivers:$rideId", $driverId);

        // ‏عدّاد العروض — بسط ومقام معدّل القبول. لا يوجد في MySQL أي سجل
        // ‏لعرضٍ لم يُقبل (driver_orders يسجّل المقبول فقط)، فالعدّ هنا هو
        // ‏المصدر الوحيد الممكن. cron_driver_score.php يحوّله لمعدّل ويضمحله.
        $redis->hincrby("driver:score:$driverId", 'offers', 1);
        $redis->zadd('dispatch:active', time() + $headStart, $rideId);

        $io->to('driver_' . $driverId)->emit('new_ride_request', $payload);
        logMsg("🎯 Ride #$rideId — سبق {$headStart}ث للسائق #$driverId");
        return;
    }

    // ‏لا مرشح متصل إطلاقاً: لا معنى لانتظار أحد — أطلق فوراً.
    logMsg("⚡ Ride #$rideId — لا مرشح متصل، إطلاق فوري للجميع");
    dispatchFallbackToBroadcast($io, $redis, $rideId);
}

/**
 * ‏الإطلاق للجميع بعد انتهاء السبق — أو فوراً إن لم يكن هناك مرشح متصل.
 * ‏هذا هو السلوك القديم نفسه، فأسوأ حالة في الوضع الجديد = الوضع القديم.
 * ‏الحمولة كاملة بالوجهة والإحداثيات — نفس ما يراه صاحب السبق.
 */
function dispatchFallbackToBroadcast($io, $redis, string $rideId): void {
    $payload = json_decode((string)$redis->get("ride:$rideId:dispatch_payload"), true) ?: [];
    dispatchCleanup($redis, $rideId);

    if (empty($payload)) {
        logMsg("⚠️  Ride #$rideId بلا حمولة — تعذّر الإطلاق");
        return;
    }

    // ‏انتهى السبق: الطلب صار عاماً للجميع. غياب الوسم في المسارات
    // ‏القديمة يعامله التطبيق كعام أيضاً، فلا حاجة لتعديلها.
    $payload['offer_type'] = 'public';
    unset($payload['offer_expires_in']);

    global $connectedDrivers;
    $count = 0;
    foreach (array_keys($connectedDrivers) as $driverId) {
        $io->to('driver_' . $driverId)->emit('market_new_ride', $payload);
        $count++;
    }
    logMsg("📢 Ride #$rideId — انتهى السبق، أُطلقت لـ $count سائق");
}

/** ‏يمسح كل مفاتيح الإسناد بالدفعات لرحلة انتهت (قُبلت أو أُلغيت أو نفدت). */
function dispatchCleanup($redis, string $rideId): void {
    if (!$redis) return;
    $redis->del("ride:$rideId:dispatch_queue");
    $redis->del("ride:$rideId:dispatch_payload");
    $redis->del("ride:$rideId:current_offer");
    $redis->zrem('dispatch:active', $rideId);
}

function loadEnvironment(string $filePath): void {
    if (!file_exists($filePath)) {
        logMsg("⚠️ .env not found: $filePath");
        return;
    }
    foreach (file($filePath, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES) as $line) {
        if (str_starts_with(trim($line), '#') || !str_contains($line, '=')) continue;
        [$name, $value] = explode('=', $line, 2);
        putenv(trim($name) . '=' . trim($value, "\"'"));
    }
    logMsg('✅ Environment loaded.');
}

// Try Docker .env first, then legacy path
$envPaths = [
    __DIR__ . '/../docker/.env',
    '/home/location/env/.env',
];
foreach ($envPaths as $envPath) {
    if (file_exists($envPath)) {
        loadEnvironment($envPath);
        break;
    }
}

// ============================================================
// 🔐 مفاتيح الأمان
// ============================================================
function getInternalSocketKey(): string {
    $key = getenv('INTERNAL_SOCKET_KEY');
    if ($key) return trim($key);
    $path = getenv('INTERNAL_SOCKET_KEY_PATH') ?: '/home/location/.internal_socket_key';
    if (file_exists($path)) return trim((string) @file_get_contents($path));
    return '';
}

$INTERNAL_KEY = getInternalSocketKey();

function getJwtSecret(): string {
    $keyPath = getenv('JWT_SECRET_KEY_PATH');
    if ($keyPath && file_exists($keyPath)) {
        return trim(file_get_contents($keyPath));
    }
    return getenv('JWT_SECRET_KEY') ?: '';
}
// Redis password: try env var, then Docker keys, then legacy path, then null (Docker Redis has no password)
$redisPass = null;
$redisPassPaths = [
    getenv('REDIS_PASS_KEY_PATH') ?: '',
    '/keys/.reds_pass_key',
    '/home/location/.reds_pass_key',
];
foreach ($redisPassPaths as $rpp) {
    if (!empty($rpp) && file_exists($rpp)) {
        $redisPass = trim((string) @file_get_contents($rpp));
        break;
    }
}

if (empty($INTERNAL_KEY)) logMsg('⚠️ Internal key not found (non-critical in Docker)');
if (empty($redisPass))    logMsg('ℹ️ Redis password not set (OK for Docker — no auth required)');

// ============================================================
// 🗄️ Redis Singleton
// ============================================================
$redis = null;

function getRedis(): ?RedisClient {
    global $redis, $redisPass;

    if ($redis !== null) {
        try {
            $redis->ping();
            return $redis;
        } catch (\Exception $e) {
            logMsg('⚠️ Redis ping failed, reconnecting...');
            $redis = null;
        }
    }

    try {
        $redisHost = getenv('REDIS_HOST') ?: ($_ENV['REDIS_HOST'] ?? (file_exists('/.dockerenv') ? 'redis' : '127.0.0.1'));
        $config = [
            'scheme'             => 'tcp',
            'host'               => $redisHost,
            'port'               => (int)(getenv('REDIS_PORT') ?: 6379),
            'read_write_timeout' => 0,
        ];
        // Only include password if it's actually set (Docker Redis has no password)
        if (!empty($redisPass)) {
            $config['password'] = $redisPass;
        }
        $client = new RedisClient($config);
        $client->connect();
        $redis = $client;
        return $redis;
    } catch (\Exception $e) {
        logMsg('❌ Redis Error: ' . $e->getMessage());
        return null;
    }
}

// ============================================================
// 📐 Haversine Distance (متر)
// ============================================================
function haversineDistance(float $lat1, float $lng1, float $lat2, float $lng2): float {
    $R    = 6371000;
    $dLat = deg2rad($lat2 - $lat1);
    $dLng = deg2rad($lng2 - $lng1);
    $a    = sin($dLat / 2) ** 2
          + cos(deg2rad($lat1)) * cos(deg2rad($lat2)) * sin($dLng / 2) ** 2;
    return $R * 2 * atan2(sqrt($a), sqrt(1 - $a));
}

// ============================================================
// 📡 Forward موقع السائق → سيرفر الراكب (ASYNC)
// ============================================================
function forwardLocationToPassengerSocket(
    string $driverId,
    string $passengerId,
    array  $payload,
    string $internalKey,
    array  &$fwdThrottle
): void {
    if (empty($passengerId)) return;

    $now    = time();
    $last   = $fwdThrottle[$driverId] ?? null;

    // ⚠️ الحمولة تستخدم latitude/longitude (انظر update_location) — قراءة
    // lat/lng هنا كانت ترجع null فتصير المسافة ضخمة، فلا يعمل الـ throttle
    // إطلاقاً ويُعاد التوجيه مع كل نبضة GPS.
    $curLat = (float)($payload['latitude'] ?? $payload['lat'] ?? 0);
    $curLng = (float)($payload['longitude'] ?? $payload['lng'] ?? 0);

    if ($last !== null) {
        $timeDiff = $now - $last['ts'];
        $dist     = haversineDistance($last['lat'], $last['lng'], $curLat, $curLng);
        if ($dist < FORWARD_MIN_METERS && $timeDiff < FORWARD_MAX_SECONDS) return;
    }

    $fwdThrottle[$driverId] = [
        'ts'  => $now,
        'lat' => $curLat,
        'lng' => $curLng,
    ];

    $passengerSocketUrl = getenv('PASSENGER_SOCKET_INTERNAL_URL') ?: 'http://127.0.0.1:3031';
    $http = new AsyncHttp();
    $http->request(
        $passengerSocketUrl,
        [
            'method'  => 'POST',
            'data'    => http_build_query([
                'action'       => 'update_driver_location',
                'passenger_id' => $passengerId,
                'payload'      => json_encode($payload),
            ]),
            'headers' => [
                'Content-Type'   => 'application/x-www-form-urlencoded',
                'x-internal-key' => $internalKey,
                'Connection'     => 'close',
            ],
            'timeout' => 3,
        ],
        null,
        fn(\Exception $e) => logMsg('⚠️ Forward failed: ' . $e->getMessage())
    );
}

// ============================================================
// 🚌 Forward موقع الباص → سيرفر الراكب (بثّ لغرفة الخط، ASYNC + throttle)
//    نفس آلية forwardLocationToPassengerSocket لكن للبثّ الجماعي
//    (باص واحد → كل ركاب الخط) بدل (سائق → راكب واحد)
// ============================================================
function forwardBusLocationToRoute(
    int    $tripId,
    int    $routeId,
    array  $payload,
    string $internalKey,
    array  &$busThrottle
): void {
    if ($routeId <= 0) return;

    $now  = time();
    $last = $busThrottle[$tripId] ?? null;

    if ($last !== null) {
        $timeDiff = $now - $last['ts'];
        $dist     = haversineDistance(
            $last['lat'], $last['lng'],
            (float)$payload['latitude'], (float)$payload['longitude']
        );
        if ($dist < FORWARD_MIN_METERS && $timeDiff < FORWARD_MAX_SECONDS) return;
    }

    $busThrottle[$tripId] = [
        'ts'  => $now,
        'lat' => (float)$payload['latitude'],
        'lng' => (float)$payload['longitude'],
    ];

    $passengerSocketUrl = getenv('PASSENGER_SOCKET_INTERNAL_URL') ?: 'http://127.0.0.1:3031';
    $http = new AsyncHttp();
    $http->request(
        $passengerSocketUrl,
        [
            'method'  => 'POST',
            'data'    => http_build_query([
                'action'   => 'broadcast_bus_location',
                'route_id' => $routeId,
                'payload'  => json_encode($payload),
            ]),
            'headers' => [
                'Content-Type'   => 'application/x-www-form-urlencoded',
                'x-internal-key' => $internalKey,
                'Connection'     => 'close',
            ],
            'timeout' => 3,
        ],
        null,
        fn(\Exception $e) => logMsg('⚠️ Bus forward failed: ' . $e->getMessage())
    );
}

// ============================================================
// 📲 FCM (ASYNC)
// ============================================================
function sendFCM_Async(string $token, string $title, string $body, array $rideData): void {
    if (empty($token)) return;

    $http = new AsyncHttp();
    $http->request(
        getenv('FCM_ENDPOINT_URL') ?: 'http://nginx/backend/ride/firebase/send_fcm.php',
        [
            'method'  => 'POST',
            'data'    => json_encode([
                'target'        => $token,
                'title'         => $title,
                'body'          => $body,
                'isTopic'       => false,
                'category'      => 'Order',
                'tone'          => 'start',
                'passengerList' => json_encode($rideData),
            ]),
            'headers' => ['Content-Type' => 'application/json; charset=UTF-8'],
            'timeout' => 5,
        ],
        null,
        fn(\Exception $e) => logMsg('⚠️ FCM failed: ' . $e->getMessage())
    );
}

// ============================================================
// 🧠 Memory State & Event Buffer
// ============================================================
$connectedDrivers      = [];
$active_orders_drivers = [];
$driverState           = [];
$fwdThrottle           = [];
$eventBuffer           = []; // 🚀 Level 2: مصفوفة تجميع الأحداث لـ Redis
$busFwdThrottle        = []; // 🚌 throttle بثّ موقع الباص لكل رحلة (trip_id)

// ============================================================
// 🚀 Socket.IO — بورت 2020
// ============================================================
$io = new SocketIO(2020);

// ============================================================
// A. Internal HTTP Server & Redis Batch Processor (Worker Start)
// ============================================================
$io->on('workerStart', function () use ($io, $INTERNAL_KEY) {

    // 🎯 0. مقدِّم طابور الإسناد بالدفعات
    // ‏مؤقّت واحد عام لكل الرحلات — لا مؤقّت لكل رحلة، فمئة رحلة متزامنة
    // ‏لا تنتج مئة مؤقّت. يمرّ كل ثانية على الرحلات التي انتهى سبقها.
    // ‏zset مرتّب بلحظة الانتهاء، فالمرور يلمس المستحقّ فقط لا كل الرحلات.
    Timer::add(1, function() use ($io) {
        $redis = getRedis();
        if (!$redis) return;

        try {
            $due = $redis->zrangebyscore('dispatch:active', 0, time());
            foreach ($due as $rideId) {
                // ‏وجود current_offer يعني أن المهلة لم تنتهِ فعلاً بعد
                // ‏(فارق تقريب بين score وTTL) — نتركها للمرور التالي.
                if ($redis->exists("ride:$rideId:current_offer")) continue;

                $redis->zrem('dispatch:active', $rideId);
                logMsg("⏱️  Ride #$rideId — انتهى السبق، إطلاق للجميع");
                dispatchFallbackToBroadcast($io, $redis, (string)$rideId);
            }
        } catch (\Exception $e) {
            logMsg('⚠️ Dispatch sweeper error: ' . $e->getMessage());
        }
    });

    // 🚀 1. Redis Pipeline Batch Processor (Level 2)
    // يعمل كل نصف ثانية، يجمع كل الأوامر ويرسلها لـ Redis دفعة واحدة
    Timer::add(REDIS_BATCH_INTERVAL, function() {
        global $eventBuffer;
        if (empty($eventBuffer)) return;

        $redis = getRedis();
        if (!$redis) return;

        try {
            $pipe = $redis->pipeline();
            $processedCount = 0;

            foreach ($eventBuffer as $driverId => $ops) {
                $profileKey = "driver:profile:$driverId";
                $processedCount++;

                if (isset($ops['hmset'])) {
                    $pipe->hmset($profileKey, $ops['hmset']);
                    // 🆕 Cache public driver data للقراءة السريعة من get.php (24h TTL)
                    $publicKey = "driver:public:$driverId";
                    $pipe->hmset($publicKey, $ops['hmset']);
                    $pipe->expire($publicKey, 86400);
                }
                if (isset($ops['expire'])) {
                    $pipe->expire($profileKey, $ops['expire']);
                }
                if (isset($ops['status_change'])) {
                    $oldStatus = $ops['status_change']['old'];
                    $newStatus = $ops['status_change']['new'];
                    
                    // إزالة من المجموعة القديمة
                    if ($oldStatus === 'on') $pipe->zrem('geo:drivers:busy', $driverId);
                    if ($oldStatus === 'off') $pipe->zrem('geo:drivers:available', $driverId);
                    
                    if ($newStatus === 'close' || $newStatus === 'blocked') {
                        $pipe->zrem('geo:drivers:available', $driverId);
                        $pipe->zrem('geo:drivers:busy', $driverId);
                    } elseif ($newStatus === 'off') {
                        // أصبح متاحاً → أضفه إلى geo:drivers:available
                        $pipe->zadd('geo:drivers:available', 0, $driverId);
                    } elseif ($newStatus === 'on') {
                        // أصبح مشغولاً → أضفه إلى geo:drivers:busy
                        $pipe->zadd('geo:drivers:busy', 0, $driverId);
                    }
                }
                if (isset($ops['geoadd'])) {
                    $st  = $ops['geoadd']['status'];
                    $lng = $ops['geoadd']['lng'];
                    $lat = $ops['geoadd']['lat'];
                    
                    if ($st === 'off') {
                        $pipe->geoadd('geo:drivers:available', $lng, $lat, $driverId);
                    } elseif ($st === 'on') {
                        $pipe->geoadd('geo:drivers:busy', $lng, $lat, $driverId);
                    }
                }
            }
            
            $pipe->execute();
            $eventBuffer = []; // إفراغ المصفوفة بعد التنفيذ الناجح

        } catch (\Exception $e) {
            logMsg("⚠️ Redis Pipeline Error: " . $e->getMessage());
        }
    });

    // 🌐 2. Internal HTTP Server — بورت 2021
    $innerHttp = new Worker('http://0.0.0.0:2021');

    $innerHttp->onMessage = function ($connection, $request) use ($io, $INTERNAL_KEY) {
        global $active_orders_drivers, $connectedDrivers;

        $headers = $request->header();
        if (($headers['x-internal-key'] ?? '') !== $INTERNAL_KEY) {
            $connection->send('Unauthorized');
            return;
        }

        $post   = $request->post();
        $action = trim($post['action'] ?? '');
        $redis  = getRedis();

        // ── 1. Dispatch Order ────────────────────────────────
        if ($action === 'dispatch_order') {
            $rideId  = $post['ride_id'] ?? null;
            $drivers = json_decode($post['drivers_ids'] ?? '[]', true);
            $payload = $post['payload'] ?? [];
            if (is_array($payload)) $payload = array_values($payload);

            if ($rideId && !empty($drivers)) {
                $active_orders_drivers[$rideId] = $drivers;
                logMsg("🚀 Dispatch Ride #$rideId → " . count($drivers) . ' drivers.');
            }

            // 🆕 Ride Offer Cache + Pending Queue (للتعافي من انقطاع الاتصال)
            if ($redis && $rideId && !empty($payload)) {
                $offerKey  = "ride:offer:$rideId";
                $offerTtl  = 120;
                $redis->setex($offerKey, $offerTtl, json_encode($payload));

                $offeredSetKey = "ride:offered_drivers:$rideId";
                $redis->sadd($offeredSetKey, ...$drivers);
                $redis->expire($offeredSetKey, $offerTtl);
            }

            foreach ($drivers as $driverId) {
                if (!isset($connectedDrivers[$driverId])) {
                    // السائق غير متصل → نخزن في قائمة الانتظار عشان يسلمله أول ما يوصل
                    if ($redis && $rideId) {
                        $pendingKey = "driver:pending_orders:$driverId";
                        $redis->zadd($pendingKey, time(), $rideId);
                        $redis->expire($pendingKey, 120);
                    }
                    logMsg("📦 Driver #$driverId offline — saved to pending queue.");
                    continue;
                }
                $io->to('driver_' . $driverId)->emit('new_ride_request', $payload);

                $platform = $connectedDrivers[$driverId]['platform'] ?? 'android';
                $token    = $connectedDrivers[$driverId]['token']    ?? '';
                if ($platform === 'ios' && !empty($token)) {
                    sendFCM_Async($token, 'طلب جديد', 'لديك رحلة جديدة قريبة منك', $payload);
                }
            }
            $connection->send('Dispatched');

        // ── 2. Market New Ride ────────────────────────────────
        } elseif ($action === 'market_new_ride') {
            $payload = $post['payload'] ?? [];
            if (is_string($payload)) {
                $decoded = json_decode($payload, true);
                if (is_array($decoded)) $payload = $decoded;
            }
            $rideId  = $payload['id']        ?? null;
            $lat     = (float)($payload['start_lat'] ?? 0);
            $lng     = (float)($payload['start_lng'] ?? 0);
            $endLat  = isset($payload['end_lat']) ? (float)$payload['end_lat'] : null;
            $endLng  = isset($payload['end_lng']) ? (float)$payload['end_lng'] : null;

            if (!$redis || !$rideId || $lat == 0 || $lng == 0) {
                $connection->send('Error: Redis unavailable or invalid coords');
                return;
            }

            $redis->geoadd('geo:rides:waiting', $lng, $lat, $rideId);
            $nearbyDrivers = $redis->georadius('geo:drivers:available', $lng, $lat, 50, 'km');

            $count = 0;
            foreach ($nearbyDrivers as $driverId) {
                if (isset($connectedDrivers[$driverId])) {
                    // Check if driver has a destination constraint in Redis
                    $profileKey = "driver:profile:$driverId";
                    $profile = $redis->hgetall($profileKey);
                    if ($profile && isset($profile['has_destination']) && $profile['has_destination'] == 1 && $endLat !== null && $endLng !== null) {
                        $driverDestLat = (float)($profile['destination_lat'] ?? 0);
                        $driverDestLng = (float)($profile['destination_lng'] ?? 0);
                        
                        $destDistance = haversineDistance($endLat, $endLng, $driverDestLat, $driverDestLng);
                        // Filter out driver if destination is > 5km (5000 meters) away
                        if ($destDistance > 5000.0) {
                            continue;
                        }
                    }

                    $io->to('driver_' . $driverId)->emit('market_new_ride', $payload);
                    $count++;
                }
            }
            logMsg("📢 Market Ride #$rideId → $count drivers.");
            $connection->send("Broadcasted to $count drivers");

        // ── 3. Get Nearby Ride IDs ────────────────────────────
        } elseif ($action === 'get_nearby_ride_ids') {
            $lat    = (float)($post['lat']    ?? 0);
            $lng    = (float)($post['lng']    ?? 0);
            $radius = (float)($post['radius'] ?? 9);

            if (!$redis) { $connection->send(json_encode([])); return; }

            $results = $redis->georadius(
                'geo:rides:waiting', $lng, $lat, $radius, 'km',
                ['WITHDIST' => true, 'SORT' => 'ASC', 'COUNT' => 40]
            );
            $connection->send(json_encode($results));

        // ── 3c. Start Batched Dispatch ────────────────────────
        //   يبنيه add_ride.php في Redis ثم يوقظنا هنا. نحن أصحاب المؤقّت
        //   لأننا العملية الوحيدة طويلة العمر — كرون كل دقيقة لا يخدم
        //   مهلة عشر ثوانٍ.
        } elseif ($action === 'start_batched_dispatch') {
            $rideId = $post['ride_id'] ?? null;
            if (!$rideId) { $connection->send('Error: Missing ride_id'); return; }

            dispatchOfferNext($io, (string)$rideId);
            $connection->send('OK');

        // ── 4. Ride Taken ─────────────────────────────────────
        } elseif ($action === 'ride_taken_event') {
            $rideId         = $post['ride_id']            ?? null;
            $winnerDriverId = $post['taken_by_driver_id'] ?? null;

            if (!$rideId) { $connection->send('Error: Missing ride_id'); return; }

            if ($redis) {
                $redis->zrem('geo:rides:waiting', $rideId);

                // 🆕 Clean pending offers from all drivers' queues
                $offeredSetKey = "ride:offered_drivers:$rideId";
                $offeredDrivers = $redis->smembers($offeredSetKey);
                foreach ($offeredDrivers as $dId) {
                    $redis->zrem("driver:pending_orders:$dId", $rideId);
                }
                $redis->del($offeredSetKey);
                $redis->del("ride:offer:$rideId");
                dispatchCleanup($redis, (string)$rideId);
            }

            $io->emit('ride_taken', [
                'ride_id'            => $rideId,
                'taken_by_driver_id' => $winnerDriverId,
            ]);

            unset($active_orders_drivers[$rideId]);
            logMsg("✅ Ride #$rideId taken by #$winnerDriverId.");
            $connection->send('OK');

        // ── 4b. Ride Cancelled (بالراكب أو بالسائق) ─────────────
        } elseif ($action === 'cancel_ride') {
            $rideId   = $post['ride_id']   ?? null;
            $driverId = $post['driver_id'] ?? null;
            $reason   = $post['reason']    ?? '';

            if (!$rideId) { $connection->send('Error: Missing ride_id'); return; }

            // كل السائقين الذين وصلهم عرض هذه الرحلة (سواء قَبِلها أحد أم لا)
            $offeredDrivers = [];

            if ($redis) {
                $redis->zrem('geo:rides:waiting', $rideId);

                $offeredSetKey  = "ride:offered_drivers:$rideId";
                $offeredDrivers = $redis->smembers($offeredSetKey);
                foreach ($offeredDrivers as $dId) {
                    $redis->zrem("driver:pending_orders:$dId", $rideId);
                }
                $redis->del($offeredSetKey);
                $redis->del("ride:offer:$rideId");
                dispatchCleanup($redis, (string)$rideId);
            }

            // نوحّد قائمة من يجب إشعارهم: السائق الحالي (إن وُجد) + كل من عُرضت عليهم الرحلة
            $notifyDrivers = $offeredDrivers;
            if ($driverId) $notifyDrivers[] = (string)$driverId;
            $notifyDrivers = array_unique(array_filter($notifyDrivers, fn($d) => $d && $d !== '0'));

            $notified = 0;
            foreach ($notifyDrivers as $dId) {
                if (isset($connectedDrivers[$dId])) {
                    $io->to('driver_' . $dId)->emit('ride_cancelled', [
                        'ride_id' => $rideId,
                        'reason'  => $reason,
                    ]);
                    $notified++;
                }
            }

            unset($active_orders_drivers[$rideId]);
            logMsg("🚫 Ride #$rideId cancelled — notified $notified/" . count($notifyDrivers) . " driver(s).");
            $connection->send('OK');

        // ── 7. Update Ride State (Redis Cache فقط — بدون Forward)
        } elseif ($action === 'update_ride_state') {
            $rideId      = $post['ride_id']      ?? null;
            $status      = $post['status']       ?? '';
            $driverId    = $post['driver_id']    ?? '';
            $passengerId = $post['passenger_id'] ?? '';

            if (!$rideId || !$status) {
                $connection->send('Error: Missing ride_id or status');
                return;
            }

            if ($redis) {
                $stateKey = "ride:$rideId:state";
                $stateData = [
                    'status'       => $status,
                    'driver_id'    => $driverId,
                    'passenger_id' => $passengerId,
                    'updated_at'   => time(),
                ];
                $redis->hmset($stateKey, $stateData);
                $redis->expire($stateKey, 86400);

                logMsg("🚗 Ride #$rideId → status: $status (cached in Redis)");
            }

            $connection->send('OK');

        // ── 7b. Get Ride State (Redis Cache Read-Only) ─────────
        //   Used by ride_server/intaleq/ride/rides/getRideStatus.php to
        //   answer passenger polling without hitting MySQL on every call.
        } elseif ($action === 'get_ride_state') {
            $rideId = (int)($post['ride_id'] ?? 0);

            if ($rideId <= 0 || !$redis) {
                $connection->send(json_encode(['status' => false, 'data' => null]));
                return;
            }

            $stateData = $redis->hgetall("ride:{$rideId}:state");

            $connection->send(json_encode([
                'status' => !empty($stateData),
                'data'   => $stateData ?: null,
            ]));

        // ── 5. Force Disconnect ───────────────────────────────
        } elseif ($action === 'force_disconnect') {
            $driverId = $post['driver_id'] ?? null;

            if ($driverId && isset($connectedDrivers[$driverId])) {
                $connectedDrivers[$driverId]['conn']->disconnect();
                unset($connectedDrivers[$driverId]);

                if ($redis) {
                    $redis->zrem('geo:drivers:available', $driverId);
                    $redis->zrem('geo:drivers:busy', $driverId);
                }
                logMsg("🚫 Driver #$driverId force-disconnected.");
                $connection->send('Disconnected');
            } else {
                $connection->send('Driver not connected');
            }

        // ── 6. Update Driver Destination ──────────────────────
        } elseif ($action === 'update_driver_destination') {
            $driverId = $post['driver_id'] ?? null;
            $hasDest  = isset($post['has_destination']) ? intval($post['has_destination']) : 0;

            if (!$driverId || !$redis) {
                $connection->send('Error: Missing driver_id or Redis unavailable');
                return;
            }

            $profileKey = "driver:profile:$driverId";
            if ($hasDest === 1) {
                $destLat  = $post['destination_lat'] ?? '';
                $destLng  = $post['destination_lng'] ?? '';
                $destName = $post['destination_name'] ?? '';

                $redis->hmset($profileKey, [
                    'has_destination'  => 1,
                    'destination_lat'  => $destLat,
                    'destination_lng'  => $destLng,
                    'destination_name' => $destName
                ]);
                $redis->expire($profileKey, 86400); // 24 Hours
                logMsg("🎯 Destination set for Driver #$driverId: $destName ($destLat, $destLng)");
            } else {
                $redis->hmset($profileKey, ['has_destination' => 0]);
                $redis->hdel($profileKey, ['destination_lat', 'destination_lng', 'destination_name']);
                logMsg("🎯 Destination cleared for Driver #$driverId");
            }
            $connection->send('OK');

        // ── 7. Cache Driver Public Data ─────────────────────────
        } elseif ($action === 'cache_driver_public') {
            $driverId = $post['driver_id'] ?? null;
            $data     = json_decode($post['data'] ?? '[]', true);

            if ($driverId && $redis && !empty($data)) {
                $key = "driver:public:$driverId";
                $redis->hmset($key, $data);
                $redis->expire($key, 86400);
                $connection->send('OK');
            } else {
                $connection->send('Error');
            }

        // ── 8. 🚌 Get Bus Position (آخر موقع للباص — للتحميل الأولي) ─
        //   يستدعيه الباك اند (transit/trip/live.php) ليعطي الراكب آخر
        //   موقع معروف فوراً عند فتح الخط، قبل أول بثّ حي عبر السوكت.
        } elseif ($action === 'get_bus_position') {
            $tripId = (int)($post['trip_id'] ?? 0);
            if ($tripId <= 0 || !$redis) {
                $connection->send(json_encode(['status' => false, 'data' => null]));
                return;
            }
            $pos = $redis->hgetall("transit:trip:$tripId:pos");
            $connection->send(json_encode([
                'status' => !empty($pos),
                'data'   => $pos ?: null,
            ]));

        } else {
            $connection->send('Unknown action');
        }
    };

    $innerHttp->listen();
});

// ============================================================
// B. WebSocket Events للسائقين
// ============================================================
$io->on('connection', function ($socket) use ($INTERNAL_KEY) {
    global $connectedDrivers, $driverState, $fwdThrottle, $eventBuffer, $busFwdThrottle;

    $query    = $socket->handshake['query'] ?? [];
    $driverId = $query['driver_id'] ?? null;
    $platform = $query['platform']  ?? 'android';
    $fcmToken = $query['token']     ?? ''; // FCM token
    $jwtToken = $query['jwt']       ?? ''; // JWT Token for authentication

    if (!$driverId) {
        logMsg("🚫 Connection Rejected: Missing driver_id");
        $socket->disconnect();
        return;
    }

    if (!empty($jwtToken)) {
        try {
            $secretKey = getJwtSecret();
            if (empty($secretKey)) {
                logMsg("⚠️ JWT Secret is not configured on the server!");
            } else {
                $decoded = JWT::decode($jwtToken, new Key($secretKey, 'HS256'));
                // Validate that the token belongs to this driver
                if ((string)$decoded->sub !== (string)$driverId || $decoded->role !== 'driver') {
                    logMsg("🚫 Connection Rejected: Invalid JWT for driver_id=$driverId");
                    $socket->disconnect();
                    return;
                }
            }
        } catch (\Exception $e) {
            logMsg("⚠️ JWT Verification skipped/failed for driver_id=$driverId -> " . $e->getMessage());
        }
    }

    $socket->join('driver_' . $driverId);

    // ‏نحفظ معرّف السوكيت مع السجلّ: التطبيق يفتح أكثر من اتصال للسائق نفسه
    // ‏(location_controller و background_service ومحاولات لم تُنظَّف)، وحين يُغلق
    // ‏أحدها كان معالج disconnect يمحو السائق من السجلّ كاملاً بينما البقية
    // ‏متصلة — فيعتبره dispatch_order غير متصل ويضع الطلب في طابور الانتظار،
    // ‏ويظهر في اللوج «notified 0/0 driver(s)».
    $socketId = $socket->id;
    $connectedDrivers[$driverId] = [
        'conn'     => $socket,
        'platform' => $platform,
        'token'    => $fcmToken,
        'sid'      => $socketId,
    ];

    if (!isset($driverState[$driverId])) {
        $driverState[$driverId] = [
            'lat'       => 0.0,
            'lng'       => 0.0,
            'speed'     => -999.0, 
            'heading'   => -999.0,
            'status'    => '',
            'expire_ts' => 0,
        ];
    }
    
    logMsg("✅ Driver Connected: #$driverId ($platform)");

    // 🆕 Deliver any pending orders missed during disconnection
    // نستخدم pipeline لتقليل عدد عمليات Redis (خفيف على السيرفر)
    $redis = getRedis();
    if ($redis) {
        $pendingKey = "driver:pending_orders:$driverId";
        $pendingRides = $redis->zrevrangebyscore($pendingKey, time(), time() - 120);
        if (!empty($pendingRides)) {
            $pipe = $redis->pipeline();
            foreach ($pendingRides as $pRideId) {
                $offerData = $redis->get("ride:offer:$pRideId");
                if ($offerData) {
                    $decoded = json_decode($offerData, true);
                    if ($decoded) {
                        $socket->emit('new_ride_request', $decoded);
                        logMsg("📦 Re-delivered pending Ride #$pRideId to Driver #$driverId");
                        // نزيل الرحلة من قائمة الانتظار فوراً
                        // هذا يمنع إعادة الإرسال عند إعادة الاتصال مرة أخرى (شبكة ضعيفة)
                        $pipe->zrem($pendingKey, $pRideId);
                    }
                } else {
                    $pipe->zrem($pendingKey, $pRideId);
                }
            }
            $pipe->execute();
        }
    }

    $socket->on('ping_alive', function () {
        // Socket.IO handles pong automatically
    });

    // 🆕 Client requests any pending orders missed during disconnection
    $socket->on('get_pending_orders', function () use ($driverId, $socket) {
        $redis = getRedis();
        if (!$redis) return;

        $pendingKey = "driver:pending_orders:$driverId";
        $pendingRides = $redis->zrevrangebyscore($pendingKey, time(), time() - 120);
        if (empty($pendingRides)) return;

        logMsg("📦 get_pending_orders: " . count($pendingRides) . " pending for Driver #$driverId");
        $pipe = $redis->pipeline();
        foreach ($pendingRides as $pRideId) {
            $offerData = $redis->get("ride:offer:$pRideId");
            if ($offerData) {
                $decoded = json_decode($offerData, true);
                if ($decoded) {
                    $socket->emit('new_ride_request', $decoded);
                    $pipe->zrem($pendingKey, $pRideId);
                }
            } else {
                $pipe->zrem($pendingKey, $pRideId);
            }
        }
        $pipe->execute();
    });

    $socket->on('update_location', function ($data)
        use ($driverId, $INTERNAL_KEY, &$driverState, &$fwdThrottle, &$eventBuffer)
    {
        global $connectedDrivers;

        $data = (array) $data;

        $lat         = isset($data['lat'])  ? (float)$data['lat']  : null;
        $lng         = isset($data['lng'])  ? (float)$data['lng']  : null;
        $heading     = (float)($data['heading']      ?? 0);
        $speed       = (float)($data['speed']        ?? 0);  
        $status      = (string)($data['status']      ?? 'off');
        $distance    = (float)($data['distance']     ?? 0);
        $passengerId = (string)($data['passenger_id'] ?? '');
        $rideId      = $data['ride_id'] ?? null;

        if ($lat === null || $lng === null) return;

        // ‏حرس: لو حُذفت الحالة (إغلاق سوكيت آخر لنفس السائق، أو إعادة تشغيل)
        // ‏فإن ‎&$driverState[$driverId]‎ يُنشئ مدخلاً null فتقرأ كل الأسطر التالية
        // ‏من null — وهذا مصدر تحذيرات «array offset on value of type null» في
        // ‏الأسطر 860‑872 و«Undefined array key status» في 907.
        if (!isset($driverState[$driverId]) || !is_array($driverState[$driverId])) {
            $driverState[$driverId] = [
                'lat'       => 0.0,
                'lng'       => 0.0,
                'speed'     => -999.0,
                'heading'   => -999.0,
                'status'    => '',
                'expire_ts' => 0,
            ];
        }

        $state = &$driverState[$driverId];
        $now   = time();

        // 1. Forward للراكب (ASYNC + throttle)
        if (!empty($passengerId)) {
            forwardLocationToPassengerSocket(
                $driverId, $passengerId,
                [
                    'latitude'  => $lat,
                    'longitude' => $lng,
                    'heading'   => $heading,
                    'speed'     => $speed,
                    'ride_id'   => $rideId,
                    'driver_id' => $driverId,
                ],
                $INTERNAL_KEY, $fwdThrottle
            );
        }

        // 2. حساب ماذا تغيّر لتجنب ضغط Redis
        $movedMeters = ($state['lat'] == 0.0 && $state['lng'] == 0.0)
            ? 999.0  
            : haversineDistance($state['lat'], $state['lng'], $lat, $lng);

        $didMove          = $movedMeters >= MIN_MOVE_METERS;
        $speedMs          = $speed / 3.6;
        $speedChanged     = abs($speedMs - $state['speed']) >= HMSET_SPEED_DELTA;
        $headingChanged   = abs($heading - $state['heading']) >= HMSET_HEADING_DELTA;
        $statusChanged    = ($status !== $state['status']);

        $needHmset        = $speedChanged || $headingChanged || $statusChanged;
        $needGeoadd       = $didMove;
        $needExpireRefresh = ($now - $state['expire_ts']) >= EXPIRE_REFRESH_SECONDS;

        if (!$needHmset && (!$needGeoadd && !$statusChanged) && !$needExpireRefresh) {
            return; // لم يتغير شيء مهم، تجاهل تماماً (0 عمليات Redis)
        }

        // 🚀 3. Buffering Event بدل الإرسال المباشر لـ Redis (Level 2 Magic)
        if (!isset($eventBuffer[$driverId])) {
            $eventBuffer[$driverId] = [];
        }

        if ($needHmset) {
            $eventBuffer[$driverId]['hmset'] = [
                'id' => $driverId, 'lat' => $lat, 'lng' => $lng,
                'heading' => $heading, 'speed' => $speed,
                'status' => $status, 'updated_at' => $now
            ];
            $state['speed']   = $speedMs;
            $state['heading'] = $heading;
        }

        if ($needExpireRefresh || $needHmset) {
            $eventBuffer[$driverId]['expire'] = 900;
            $state['expire_ts'] = $now;
            if (!isset($eventBuffer[$driverId]['hmset'])) {
                $eventBuffer[$driverId]['hmset'] = [
                    'id' => $driverId, 'lat' => $lat, 'lng' => $lng, 'updated_at' => $now
                ];
            } else {
                $eventBuffer[$driverId]['hmset']['updated_at'] = $now;
            }
        }

        if ($statusChanged) {
            $eventBuffer[$driverId]['status_change'] = [
                'old' => $state['status'], 
                'new' => $status
            ];
            $state['status'] = $status;
            
            // Auto disconnect if blocked
            if ($status === 'blocked') {
                if (isset($connectedDrivers[$driverId])) {
                    $connectedDrivers[$driverId]['conn']->disconnect();
                    unset($connectedDrivers[$driverId]);
                }
            }
        }

        if ($needGeoadd || $statusChanged) {
            $eventBuffer[$driverId]['geoadd'] = [
                'status' => $status,
                'lng'    => $lng,
                'lat'    => $lat
            ];
            if ($needGeoadd) {
                $state['lat'] = $lat;
                $state['lng'] = $lng;
                // 🆕 تحديث الموقع في driver:public (حتى لو ما تغير speed/heading)
                if (!isset($eventBuffer[$driverId]['hmset'])) {
                    $eventBuffer[$driverId]['hmset'] = [
                        'id' => $driverId, 'lat' => $lat, 'lng' => $lng,
                        'updated_at' => $now
                    ];
                }
            }
        }
    });

    // ── 🚌 وضع الباص (مواصلاتي) ────────────────────────────────
    //   سائق الباص هو سائق عادي (JWT role=driver) لكنه في وضع الباص لا
    //   يدخل حوض الرحلات (geo:drivers:*). يبعث update_bus_location فقط.
    //   نخزّن آخر موقع في Redis سيرفر الموقع + نبثّه لغرفة الخط.
    $socket->on('update_bus_location', function ($data)
        use ($driverId, $INTERNAL_KEY, &$busFwdThrottle)
    {
        $data = (array) $data;

        $tripId  = (int)($data['trip_id']  ?? 0);
        $lat     = isset($data['lat']) ? (float)$data['lat'] : null;
        $lng     = isset($data['lng']) ? (float)$data['lng'] : null;
        $heading = (float)($data['heading'] ?? 0);
        $speed   = (float)($data['speed']   ?? 0);
        $stopSeq = isset($data['current_stop_seq']) ? (int)$data['current_stop_seq'] : null;

        if (!$tripId || $lat === null || $lng === null) return;

        $redis = getRedis();
        if (!$redis) return;

        // ── تحقق الملكية: هذه الرحلة فعلاً مسندة لهذا السائق؟ ────────
        //   الكاش يُكتب من transit/trip/start.php عند بدء الرحلة (transitSetTripOwner)
        //   ويُحذف عند إنهائها. أي سائق آخر (حتى لو خمّن trip_id صحيح) يُرفض هنا،
        //   ولا نثق بـ route_id القادم من العميل — نأخذه دائماً من الكاش الموثوق.
        $owner = $redis->hgetall("transit:trip:$tripId:owner");
        if (empty($owner) || (string)($owner['driver_id'] ?? '') !== (string)$driverId) {
            logMsg("🚫 update_bus_location rejected: driver #$driverId is not the owner of trip #$tripId");
            return;
        }
        $routeId = (int)($owner['route_id'] ?? 0);
        if ($routeId <= 0) return;

        // 1. آخر موقع في Redis سيرفر الموقع (بدون بادئة — يقرؤه الباك اند عبر $redisLocation)
        $posKey = "transit:trip:$tripId:pos";
        $redis->hmset($posKey, [
            'lat'       => $lat,
            'lng'       => $lng,
            'heading'   => $heading,
            'speed'     => $speed,
            'driver_id' => $driverId,
            'ts'        => time(),
        ]);
        $redis->expire($posKey, 86400);

        // 2. بثّ الموقع لكل ركاب الخط عبر سيرفر الراكب (throttled)
        forwardBusLocationToRoute($tripId, $routeId, [
            'trip_id'          => $tripId,
            'route_id'         => $routeId,
            'latitude'         => $lat,
            'longitude'        => $lng,
            'heading'          => $heading,
            'speed'            => $speed,
            'current_stop_seq' => $stopSeq,
            'driver_id'        => $driverId,
        ], $INTERNAL_KEY, $busFwdThrottle);
    });

    $socket->on('disconnect', function () use ($driverId, $socketId) {
        global $connectedDrivers, $driverState, $fwdThrottle;

        // ‏لا ننظّف إلا إذا كان المُغلَق هو السوكيت المسجَّل حالياً. الجهاز يفتح
        // ‏عدة اتصالات للسائق نفسه، وتنظيف السجلّ عند إغلاق نسخة قديمة كان
        // ‏يُطفئ السائق فعلياً وهو متصل: التوزيع يراه offline، و update_location
        // ‏من السوكيت الحيّ يجد $driverState محذوفاً فتظهر تحذيرات null.
        $registeredSid = $connectedDrivers[$driverId]['sid'] ?? null;
        if ($registeredSid !== null && $registeredSid !== $socketId) {
            logMsg("↩️ Stale duplicate socket closed for #$driverId — registry kept.");
            return;
        }

        unset($connectedDrivers[$driverId]);
        unset($driverState[$driverId]);
        unset($fwdThrottle[$driverId]);
        // ملاحظة: $busFwdThrottle مُفهرس بـ trip_id (لا driver_id) — يُستبدَل
        // تلقائياً في الرحلة التالية، فلا حاجة لحذفه هنا.

        logMsg("❌ Driver Disconnected: #$driverId");
    });
});

Worker::runAll();