Update: 2026-08-07 01:35:27

loction_server/driver_socket.php

@@ -42,6 +42,114 @@ 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");
@@ -315,6 +423,30 @@ $io = new SocketIO(2020);
 // ============================================================
 $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() {
@@ -501,6 +633,17 @@ $io->on('workerStart', function () use ($io, $INTERNAL_KEY) {
             );
             $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;
@@ -519,6 +662,7 @@ $io->on('workerStart', function () use ($io, $INTERNAL_KEY) {
                 }
                 $redis->del($offeredSetKey);
                 $redis->del("ride:offer:$rideId");
+                dispatchCleanup($redis, (string)$rideId);
             }
 
             $io->emit('ride_taken', [
@@ -551,6 +695,7 @@ $io->on('workerStart', function () use ($io, $INTERNAL_KEY) {
                 }
                 $redis->del($offeredSetKey);
                 $redis->del("ride:offer:$rideId");
+                dispatchCleanup($redis, (string)$rideId);
             }
 
             // نوحّد قائمة من يجب إشعارهم: السائق الحالي (إن وُجد) + كل من عُرضت عليهم الرحلة