Siro/backend/ride/dispatch/dispatch_helper.php

<?php
// ============================================================
// dispatch_helper.php — طبقة القرار بين إنشاء الرحلة والسوكِت
//
// ‏الوضع القائم (broadcast): الرحلة تُبَثّ لكل سائق قريب، ويفوز أسرع من
// ‏يضغط — لا الأقرب. سباق يخسره تسعة من عشرة.
//
// ‏الوجهة تبقى ظاهرة كاملة (إحداثيات واسماً) بقرار المالك: التطبيقات ترسم
// ‏بها المسار والخريطة، وللسائق حق معرفة وجهته قبل القبول. معالجة الرفض
// ‏الانتقائي تتم عبر وزن معدّل القبول في الترتيب أدناه، لا بحجب المعلومة.
//
// ‏الوضع الجديد (batched): سبق حصري قصير للمرشح الأفضل، ثم إطلاق للجميع.
//
// ‏قرار مالك المنتج صراحةً: لا سُلّم تنازلي بين المرشحين. سُلّم من ثمانية
// ‏مرشحين بمهلة عشر ثوانٍ لكلٍّ يعني انتظاراً قد يبلغ ثمانين ثانية للراكب —
// ‏وانتظار الراكب هو المؤشر الذي يقتل المنتج، لا عدالة التوزيع بين
// ‏السائقين. التأخير هنا محدود بخمس ثوانٍ ثابتة ومعروفة، ثم يرى الجميع
// ‏الرحلة كما في السلوك القديم.
//
// ⚠️ يلمس أخطر مسار في النظام. لذلك:
//   • مطفأ افتراضياً — DISPATCH_MODE غير المضبوط = broadcast (السلوك الحالي حرفياً)
//   • قابل للحصر بمدينة/نوع سيارة عبر DISPATCH_BATCHED_CAR_TYPES
//   • التراجع فوري: غيّر متغير البيئة وأعد تشغيل php — بلا ترحيل ولا نشر
//
// ‏لا يضيف هذا الملف أي استعلام MySQL: كل إشارات الترتيب تأتي من Redis أو
// ‏من بيانات findBestDrivers() المحمّلة أصلاً. الإشارة الغائبة تُحسب محايدة.
// ============================================================

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

// ‏عدد المرشحين المحفوظين. ليس سُلّماً: يُستخدم فقط لاختيار صاحب السبق إن
// ‏كان الأول غير متصل لحظتها — الانتقال بينهم فوري بلا أي انتظار إضافي.
const DISPATCH_MAX_CANDIDATES = 3;

/**
 * هل نستخدم الإسناد بالدفعات لهذه الرحلة؟
 *
 * DISPATCH_MODE=batched يفعّله، وأي قيمة أخرى (أو غيابها) تبقي البثّ الحر.
 * DISPATCH_BATCHED_CAR_TYPES يحصره بأنواع سيارات بعينها للتجربة التدريجية،
 * مثلاً "Speed" لتجريبه على نوع واحد قبل التعميم. فارغ = كل الأنواع.
 */
function dispatchIsBatched(string $carType): bool
{
    if (strtolower((string) getenv('DISPATCH_MODE')) !== 'batched') {
        return false;
    }

    $allowed = trim((string) getenv('DISPATCH_BATCHED_CAR_TYPES'));
    if ($allowed === '') {
        return true;
    }

    $list = array_filter(array_map('trim', explode(',', $allowed)));
    return in_array($carType, $list, true);
}

/**
 * درجة المرشح — كلما ارتفعت كان أولى بالعرض.
 *
 * الأوزان مقصودة كنقطة انطلاق قابلة للمعايرة بعد أول أسبوع تشغيل، لا
 * كأرقام نهائية. كل مكوّن مُطبَّع إلى [0,1] فيبقى المجموع مفهوماً.
 *
 * @param array $driver صف من findBestDrivers()
 * @param array $signals إشارات مقروءة من Redis (قد تكون فارغة)
 */
function dispatchScoreDriver(array $driver, array $signals): float
{
    // ── ١. القرب (40%) — بديل عملي عن زمن الوصول الفعلي.
    // ‏المسافة الهوائية ليست ETA حقيقياً، لكنها الإشارة الوحيدة المتاحة بلا
    // ‏نداء OSRM لكل مرشح. ترقية هذا إلى ETA حقيقي هي أول تحسين بعد الإطلاق.
    $distanceKm = (float) ($driver['distance_km'] ?? 999);
    $proximity  = $distanceKm >= 10.0 ? 0.0 : (1.0 - ($distanceKm / 10.0));

    // ── ٢. معدّل القبول التاريخي (20%) — من Redis، محايد إن غاب.
    $acceptRate = isset($signals['accept_rate'])
        ? max(0.0, min(1.0, (float) $signals['accept_rate']))
        : 0.5;

    // ── ٣. التقييم (15%) — من ٥، محايد إن غاب.
    $rating = isset($signals['rating'])
        ? max(0.0, min(1.0, ((float) $signals['rating']) / 5.0))
        : 0.5;

    // ── ٤. مدة الخمول (15%) — إنصاف: من انتظر أطول يُقدَّم.
    // ‏يُشبَع عند ٣٠ دقيقة حتى لا يحتكر سائق نائم كل الرحلات.
    $idleSeconds = (int) ($signals['idle_seconds'] ?? 0);
    $idle = $idleSeconds <= 0 ? 0.5 : min(1.0, $idleSeconds / 1800.0);

    // ── ٥. ملاءمة الوجهة (10%) — السائق الذي سجّل وجهة توافق مسار الرحلة.
    // ‏findBestDrivers يستبعد من وجهته أبعد من ٥كم، فوجود الوجهة هنا يعني
    // ‏توافقاً مؤكداً لا مجرد احتمال.
    $destinationFit = !empty($driver['has_destination']) ? 1.0 : 0.5;

    $score = (0.40 * $proximity)
        + (0.20 * $acceptRate)
        + (0.15 * $rating)
        + (0.15 * $idle)
        + (0.10 * $destinationFit);

    // ── عقوبة التسرب خارج المنصة ──
    // ‏يحسبها bot/cron_leak_detection.php من نمط: قَبِل ← أُلغيت ← لم تبدأ،
    // ‏مع تكرار نفس زوج (سائق، راكب). ليست حظراً بل خفض أولوية: السائق
    // ‏يخسر السبق والرحلات الأقرب تدريجياً — موجع بلا حرمان، ويتراجع من
    // ‏نفسه حين ينظف سجله لأن الكرون يحذف الحقل خارج النافذة.
    //
    // ‏تُطرح من الدرجة لا تُضرب فيها: الضرب كان سيبقي السائق الأقرب
    // ‏متقدّماً رغم عقوبته، والطرح يزيحه فعلاً عن رأس القائمة.
    $leakPenalty = isset($signals['leak_penalty'])
        ? max(0.0, min(1.0, (float) $signals['leak_penalty']))
        : 0.0;

    return round(max(0.0, $score - (0.35 * $leakPenalty)), 4);
}

/**
 * يقرأ إشارات الترتيب لكل السائقين المرشحين بنداء Redis واحد (pipeline).
 *
 * ‏المفاتيح تُغذّى من كرونات قائمة أو تُترك فارغة — الغياب لا يكسر شيئاً،
 * ‏بل يجعل المكوّن محايداً. هذا يسمح بإطلاق المحرك قبل بناء مغذّياته.
 */
function dispatchReadSignals($redisLocation, array $driverIds): array
{
    $signals = [];
    if (!$redisLocation || empty($driverIds)) {
        return $signals;
    }

    try {
        $pipe = $redisLocation->multi(Redis::PIPELINE);
        foreach ($driverIds as $id) {
            $pipe->hGetAll("driver:score:$id");
        }
        $rows = $pipe->exec() ?: [];

        foreach ($driverIds as $i => $id) {
            $row = $rows[$i] ?? [];
            if (!is_array($row) || empty($row)) {
                continue;
            }
            $signals[$id] = [
                'accept_rate' => $row['accept_rate'] ?? null,
                'rating'      => $row['rating']      ?? null,
                // ‏آخر لحظة أنهى فيها رحلة — نحوّلها لمدة خمول.
                'idle_seconds' => isset($row['last_ride_end'])
                    ? max(0, time() - (int) $row['last_ride_end'])
                    : 0,
                'leak_penalty' => $row['leak_penalty'] ?? null,
            ];
        }
    } catch (Throwable $e) {
        error_log("[dispatch] تعذّرت قراءة إشارات الترتيب: " . $e->getMessage());
    }

    return $signals;
}

/**
 * ‏أول مرشح في الطابور — أي صاحب السبق المتوقّع.
 *
 * ‏يُقرأ بلا إخراج (lIndex لا lPop): السوكِت هو من يسحب من الطابور، وسحبنا
 * ‏منه هنا كان سيحرم الأول من عرضه.
 *
 * ⚠️ ‏قد يتخطّاه السوكِت إن كان غير متصل، فيصير السبق لغيره بينما وصل هذا
 * ‏إشعار FCM. لا ضرر: تطبيقه سيجد الرحلة مأخوذة أو عامة، والسبق ينتهي
 * ‏بعد ثوانٍ على أي حال.
 */
function dispatchHeadStartDriver($redisLocation, $rideId): ?string
{
    if (!$redisLocation) {
        return null;
    }
    try {
        $first = $redisLocation->lIndex("ride:$rideId:dispatch_queue", 0);
        return ($first === false || $first === null) ? null : (string) $first;
    } catch (Throwable $e) {
        error_log("[dispatch] تعذّرت قراءة صاحب السبق RideID=$rideId: " . $e->getMessage());
        return null;
    }
}

/**
 * ‏يرتّب المرشحين ويسلّمهم للسوكِت ليمنح الأفضلَ سبقاً حصرياً قصيراً.
 *
 * @return bool ‏true إن بدأ الإسناد بالدفعات، false إن تعذّر (فيجب أن يقع
 *              ‏المستدعي على البثّ الحر — لا تُترك رحلة بلا إسناد).
 */
function dispatchStartBatched(
    $redisLocation,
    $rideId,
    array $driversData,
    array $marketPayload
): bool {
    if (!$redisLocation || empty($driversData)) {
        error_log("[dispatch] لا مرشحين أو Redis غير متاح RideID=$rideId"
            . " — السقوط للبثّ الحر");
        return false;
    }

    $driverIds = array_values(array_filter(array_column($driversData, 'driver_id')));
    $signals   = dispatchReadSignals($redisLocation, $driverIds);

    // الترتيب بالدرجة تنازلياً
    $ranked = [];
    foreach ($driversData as $driver) {
        $id = $driver['driver_id'] ?? null;
        if (!$id) {
            continue;
        }
        $ranked[] = [
            'driver_id' => (string) $id,
            'score'     => dispatchScoreDriver($driver, $signals[$id] ?? []),
        ];
    }

    usort($ranked, fn($a, $b) => $b['score'] <=> $a['score']);
    $ranked = array_slice($ranked, 0, DISPATCH_MAX_CANDIDATES);

    if (empty($ranked)) {
        return false;
    }

    try {
        $queueKey = "ride:$rideId:dispatch_queue";
        $redisLocation->del($queueKey);
        foreach ($ranked as $r) {
            $redisLocation->rPush($queueKey, $r['driver_id']);
        }
        // ‏عمر قصير: السبق خمس ثوانٍ ثم يُطلق للجميع، فلا معنى لبقاء
        // ‏الطابور بعدها. هامش سخي للمهل الشبكية فقط.
        $ttl = dispatchHeadStartSeconds() + 30;
        $redisLocation->expire($queueKey, $ttl);

        // ‏الحمولة التي سيراها المرشح — بلا وجهة نهائية (انظر أدناه).
        $redisLocation->setex(
            "ride:$rideId:dispatch_payload",
            $ttl,
            // ‏الحمولة كاملة بالإحداثيات واسم الوجهة — قرار المالك: السائق
            // ‏يحتاجها لرسم البولي‌لاين والخريطة، ومن حقه معرفة وجهته قبل
            // ‏القبول (منطقة خطرة، بعيدة، أو لا يريدها). لا إخفاء.
            json_encode($marketPayload, JSON_UNESCAPED_UNICODE)
        );
    } catch (Throwable $e) {
        error_log("[dispatch] تعذّر بناء الطابور RideID=$rideId: " . $e->getMessage());
        return false;
    }

    error_log("[dispatch] RideID=$rideId — سبق " . dispatchHeadStartSeconds()
        . "ث للسائق " . $ranked[0]['driver_id']
        . " (درجة " . $ranked[0]['score'] . ") ثم إطلاق للجميع");

    return true;
}