971 lines
40 KiB
PHP
Executable File
971 lines
40 KiB
PHP
Executable File
<?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 = 300; // 5 دقائق لتجديد الـ TTL
|
|
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;
|
|
}
|
|
|
|
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.');
|
|
}
|
|
|
|
loadEnvironment('/home/location/env/.env');
|
|
|
|
// ============================================================
|
|
// 🔐 مفاتيح الأمان
|
|
// ============================================================
|
|
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') ?: '';
|
|
}
|
|
$redisPass = trim((string) @file_get_contents('/home/location/.reds_pass_key'));
|
|
|
|
if (empty($INTERNAL_KEY)) logMsg('❌ CRITICAL: Internal key missing!');
|
|
if (empty($redisPass)) logMsg('❌ CRITICAL: Redis password missing!');
|
|
|
|
// ============================================================
|
|
// 🗄️ 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 {
|
|
$client = new RedisClient([
|
|
'scheme' => 'tcp',
|
|
'host' => '127.0.0.1',
|
|
'port' => 6379,
|
|
'password' => $redisPass,
|
|
'read_write_timeout' => 0,
|
|
]);
|
|
$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;
|
|
|
|
if ($last !== null) {
|
|
$timeDiff = $now - $last['ts'];
|
|
$dist = haversineDistance(
|
|
$last['lat'], $last['lng'],
|
|
(float)$payload['lat'], (float)$payload['lng']
|
|
);
|
|
if ($dist < FORWARD_MIN_METERS && $timeDiff < FORWARD_MAX_SECONDS) return;
|
|
}
|
|
|
|
$fwdThrottle[$driverId] = [
|
|
'ts' => $now,
|
|
'lat' => (float)$payload['lat'],
|
|
'lng' => (float)$payload['lng'],
|
|
];
|
|
|
|
$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(
|
|
'https://api.intaleq.xyz/intaleq/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) {
|
|
|
|
// 🚀 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'] ?? [];
|
|
$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));
|
|
|
|
// ── 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");
|
|
}
|
|
|
|
$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");
|
|
}
|
|
|
|
// نوحّد قائمة من يجب إشعارهم: السائق الحالي (إن وُجد) + كل من عُرضت عليهم الرحلة
|
|
$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 || empty($jwtToken)) {
|
|
logMsg("🚫 Connection Rejected: Missing driver_id or jwt token");
|
|
$socket->disconnect();
|
|
return;
|
|
}
|
|
|
|
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("🚫 Connection Rejected: JWT Verification failed -> " . $e->getMessage());
|
|
$socket->disconnect();
|
|
return;
|
|
}
|
|
|
|
$socket->join('driver_' . $driverId);
|
|
$connectedDrivers[$driverId] = [
|
|
'conn' => $socket,
|
|
'platform' => $platform,
|
|
'token' => $fcmToken,
|
|
];
|
|
|
|
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;
|
|
|
|
$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 ($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) {
|
|
global $connectedDrivers, $driverState, $fwdThrottle;
|
|
|
|
unset($connectedDrivers[$driverId]);
|
|
unset($driverState[$driverId]);
|
|
unset($fwdThrottle[$driverId]);
|
|
// ملاحظة: $busFwdThrottle مُفهرس بـ trip_id (لا driver_id) — يُستبدَل
|
|
// تلقائياً في الرحلة التالية، فلا حاجة لحذفه هنا.
|
|
|
|
logMsg("❌ Driver Disconnected: #$driverId");
|
|
});
|
|
});
|
|
|
|
Worker::runAll(); |