Files
intaleq/loction_server/driver_socket.php
T

1064 lines
46 KiB
PHP
Executable File
Raw Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
<?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;
}
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) {
// 🚀 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));
// ── 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) {
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->user_id !== (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,
];
} else {
// ‏اتصال جديد لا يضمن أن السائق ما زال عضواً في geo:drivers:* —
// ‏api_get_nearby.php يحذف العضوية كلّما وجد driver:profile منتهياً
// ‏(TTL = 900 ثانية)، بينما تبقى هذه الحالة في ذاكرة الـ worker لأن
// ‏سوكيت خدمة الخلفية (background_service) يظل مسجَّلاً فيمنع تنظيفها
// ‏عند إغلاق سوكيت الواجهة.
// ‏وبلا تصفيرها يجد أول update_location بعد فتح التطبيق أن status و
// ‏lat/lng بلا تغيير ⇒ لا GEOADD ⇒ يبقى الكابتن غير مرئي على خريطة
// ‏الراكب حتى يتحرك 10 أمتار أو يضغط زر التوفّر (الذي يقطع السوكيت).
// ‏نصفّر الجزء المتطاير فقط ليُجبَر أول تحديث على إعادة تسجيله.
$driverState[$driverId]['lat'] = 0.0;
$driverState[$driverId]['lng'] = 0.0;
$driverState[$driverId]['status'] = '';
$driverState[$driverId]['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]);
}
}
}
// ‏نُعيد تأكيد العضوية في geo:drivers:* مع كل تجديد صلاحية (كل
// ‏EXPIRE_REFRESH_SECONDS) لا عند الحركة فقط: الكابتن المتاح والواقف
// ‏لا يُصدر didMove ولا statusChanged أبداً، فلو حُذف من المجموعة
// ‏(انقطاع، إعادة تشغيل السيرفر، تنظيف api_get_nearby) لما عاد إليها
// ‏قط. التكلفة: عملية GEOADD واحدة لكل كابتن كل دقيقتين.
if ($needGeoadd || $statusChanged || $needExpireRefresh) {
$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();