Siro/ride_server/passenger_socket.php

<?php
/**
 * passenger_socket.php
 * =====================
 * WebSocket Server للركاب   — بورت 3030
 * Internal HTTP Server      — بورت 3031
 */

use Workerman\Worker;
use PHPSocketIO\SocketIO;
use Firebase\JWT\JWT;
use Firebase\JWT\Key;

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

// ---------------------------------------------------------
// نظام تسجيل الأحداث (Logging System)
// ---------------------------------------------------------
$LOG_FILE = __DIR__ . '/socket_debug.log';

function socket_log($message, $data = null) {
    global $LOG_FILE;
    $date = date('Y-m-d H:i:s');
    $logMsg = "[$date] $message";
    if ($data !== null) {
        $logMsg .= " | DATA: " . (is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_UNICODE));
    }
    $logMsg .= PHP_EOL;
    
    echo $logMsg; // للطباعة في الكونسول إذا كان يعمل في الـ Foreground
    @file_put_contents($LOG_FILE, $logMsg, FILE_APPEND); // الكتابة في الملف
}

// ---------------------------------------------------------

socket_log("=== STARTING PASSENGER SOCKET SERVER ===");

function loadEnvironment(string $filePath): void {
    if (!file_exists($filePath)) {
        socket_log("[WARNING] .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, "\"'"));
    }
}
loadEnvironment(dirname(__DIR__) . '/docker/.env');
loadEnvironment(dirname(__DIR__) . '/backend/.env');
loadEnvironment('/home/intaleq-rides/env/.env');

function getInternalSocketKey(): string {
    $key = getenv('INTERNAL_SOCKET_KEY');
    if ($key) return trim($key);
    $path = getenv('INTERNAL_SOCKET_KEY_PATH');
    if ($path && file_exists($path)) return trim((string) @file_get_contents($path));
    if (file_exists('/keys/internal_socket_key')) return trim((string) @file_get_contents('/keys/internal_socket_key'));
    if (file_exists('/home/location/.internal_socket_key')) return trim((string) @file_get_contents('/home/location/.internal_socket_key'));
    if (file_exists('/home/intaleq-rides/.internal_socket_key')) return trim((string) @file_get_contents('/home/intaleq-rides/.internal_socket_key'));
    return '';
}

$INTERNAL_KEY = getInternalSocketKey();

// ── Redis سيرفر الموقع (للتحقق من عضوية الراكب في خط مواصلاتي) ──
//   نفس الـ Redis الذي يكتب عليه driver_socket و transit/functions.php.
//   الهوست من REDIS_HOST حتى يعمل داخل Docker وخارجه على حدّ سواء.
$_locationRedis = null;

function getLocationRedis(): ?\Redis {
    global $_locationRedis;
    if ($_locationRedis !== null) {
        try { $_locationRedis->ping(); return $_locationRedis; }
        catch (\Exception $_) { $_locationRedis = null; }
    }
    try {
        $redisPass = '';
        foreach ([getenv('REDIS_PASS_KEY_PATH') ?: '', '/keys/.reds_pass_key', '/home/location/.reds_pass_key'] as $p) {
            if (!empty($p) && file_exists($p)) { $redisPass = trim((string)@file_get_contents($p)); break; }
        }
        $host = getenv('REDIS_HOST') ?: (file_exists('/.dockerenv') ? 'redis' : '127.0.0.1');
        $r = new \Redis();
        $r->connect($host, (int)(getenv('REDIS_PORT') ?: 6379), 1.5);
        if ($redisPass) $r->auth($redisPass);
        $_locationRedis = $r;
        return $r;
    } catch (\Exception $e) {
        socket_log("[REDIS_ERROR] Location Redis unavailable: " . $e->getMessage());
        return null;
    }
}

function getJwtSecret(): string {
    $keyPath = getenv('JWT_SECRET_KEY_PATH');
    if ($keyPath && file_exists($keyPath)) {
        return trim(file_get_contents($keyPath));
    }
    if (file_exists('/keys/jwt_secret_key')) {
        return trim(file_get_contents('/keys/jwt_secret_key'));
    }
    return getenv('JWT_SECRET_KEY') ?: (getenv('JWT_SECRET') ?: '');
}

if (empty($INTERNAL_KEY)) {
    socket_log("[CRITICAL_ERROR] Internal key missing! Exiting.");
    exit(1);
}

$PORT          = 3030;
$INTERNAL_PORT = 3031;

$io = new SocketIO($PORT);

$io->on('workerStart', function () use ($io, $INTERNAL_KEY, $INTERNAL_PORT) {

    $innerHttp = new Worker("http://0.0.0.0:$INTERNAL_PORT");

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

        $headers = $request->header();
        $clientIp = $connection->getRemoteIp();
        
        if (($headers['x-internal-key'] ?? '') !== $INTERNAL_KEY) {
            socket_log("[HTTP_ERROR] Unauthorized internal request from IP: $clientIp");
            $connection->send('Unauthorized');
            return;
        }

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

        if ($action === 'update_ride_status') {

            $passengerId = $post['passenger_id'] ?? null;
            $rawPayload  = $post['payload']       ?? null;

            if (!$passengerId || !$rawPayload) {
                socket_log("[HTTP_ERROR] Missing passenger_id or payload for action: update_ride_status", $post);
                $connection->send('Error: Missing passenger_id or payload');
                return;
            }

            $payload = is_string($rawPayload)
                ? (json_decode($rawPayload, true) ?? $rawPayload)
                : $rawPayload;

            socket_log("[HTTP_SUCCESS] Emitting 'ride_status_change' to Passenger #$passengerId", $payload);
            $io->to('passenger_' . $passengerId)->emit('ride_status_change', $payload);

            $connection->send('OK');

        } elseif ($action === 'update_driver_location') {

            $passengerId = $post['passenger_id'] ?? null;
            $rawPayload  = $post['payload']       ?? null;

            if (!$passengerId || !$rawPayload) {
                socket_log("[HTTP_ERROR] Missing passenger_id or payload for action: update_driver_location", $post);
                $connection->send('Error: Missing passenger_id or payload');
                return;
            }

            $payload = is_string($rawPayload)
                ? (json_decode($rawPayload, true) ?? $rawPayload)
                : $rawPayload;

            socket_log("[HTTP_SUCCESS] Emitting 'driver_location_update' to Passenger #$passengerId", $payload);
            $io->to('passenger_' . $passengerId)->emit('driver_location_update', $payload);

            $connection->send('OK');

        } elseif ($action === 'broadcast_bus_location') {

            // 🚌 بثّ موقع الباص لكل ركاب الخط المشتركين (مواصلاتي)
            //   يصل من driver_socket بعد أن يبعث سائق الباص update_bus_location
            $routeId    = $post['route_id'] ?? null;
            $rawPayload = $post['payload']  ?? null;

            if (!$routeId || !$rawPayload) {
                socket_log("[HTTP_ERROR] Missing route_id or payload for action: broadcast_bus_location", $post);
                $connection->send('Error: Missing route_id or payload');
                return;
            }

            $payload = is_string($rawPayload)
                ? (json_decode($rawPayload, true) ?? $rawPayload)
                : $rawPayload;

            socket_log("[HTTP_SUCCESS] Emitting 'bus_location_update' to route #$routeId", $payload);
            $io->to('transit_route_' . $routeId)->emit('bus_location_update', $payload);

            $connection->send('OK');

        } else {
            socket_log("[HTTP_WARNING] Unknown action received: $action", $post);
            $connection->send('Unknown action: ' . $action);
        }
    };

    $innerHttp->listen();
    socket_log("[INFO] Internal HTTP started on port $INTERNAL_PORT");
});

$io->on('connection', function ($socket) {

    $query       = $socket->handshake['query'] ?? [];
    $passengerId = $query['id'] ?? null;
    $jwtToken    = $query['jwt'] ?? ''; // JWT Token for authentication
    $clientIp    = $socket->conn->remoteAddress ?? 'Unknown';

    if (!$passengerId || empty($jwtToken)) {
        socket_log("[SOCKET_REJECTED] Connection rejected (No passenger ID or JWT missing) from IP: $clientIp");
        $socket->disconnect();
        return;
    }

    try {
        $secretKey = getJwtSecret();
        if (empty($secretKey)) {
            socket_log("[WARNING] JWT Secret is not configured on the server!");
        } else {
            $decoded = JWT::decode($jwtToken, new Key($secretKey, 'HS256'));
            if ((string)$decoded->sub !== (string)$passengerId || $decoded->role !== 'passenger') {
                socket_log("[SOCKET_REJECTED] Connection rejected: Invalid JWT for passenger_id=$passengerId from IP: $clientIp");
                $socket->disconnect();
                return;
            }
        }
    } catch (\Exception $e) {
        socket_log("[SOCKET_REJECTED] Connection rejected: JWT Verification failed -> " . $e->getMessage() . " from IP: $clientIp");
        $socket->disconnect();
        return;
    }

    $socket->join('passenger_' . $passengerId);
    socket_log("[SOCKET_CONNECTED] Passenger Connected: #$passengerId (IP: $clientIp)");

    $socket->on('heartbeat', function ($data) {
        // يمكن تفعيل السطر التالي للتأكد من النبضات إذا أردت دقة شديدة، لكنه قد يملأ ملف الـ log
        // socket_log("[SOCKET_HEARTBEAT] Received from Passenger #$passengerId");
    });

    // 🚌 اشتراك الراكب في بثّ موقع باص خط (مواصلاتي)
    //   يُرفض إن لم يكن الراكب عضواً نشطاً في المؤسسة المالكة للخط
    $socket->on('subscribe_transit_route', function ($data) use ($socket, $passengerId) {
        $data    = (array) $data;
        $routeId = (int)($data['route_id'] ?? 0);
        if ($routeId <= 0) return;

        // transit:route_org:{routeId}   → org_id  (كُتب في route/approve.php)
        // transit:org_members:{orgId}   → SET من passenger_ids (transitCacheEnrollment)
        $redis = getLocationRedis();
        if ($redis) {
            $orgId = $redis->get("transit:route_org:{$routeId}");
            if ($orgId === false || $orgId === null) {
                socket_log("[TRANSIT_BLOCKED] Route #{$routeId} not in Redis — passenger #{$passengerId}");
                $socket->emit('transit_error', ['code' => 'ROUTE_NOT_FOUND', 'route_id' => $routeId]);
                return;
            }
            if (!$redis->sIsMember("transit:org_members:{$orgId}", (string)$passengerId)) {
                socket_log("[TRANSIT_BLOCKED] Passenger #{$passengerId} not enrolled in org #{$orgId} (route #{$routeId})");
                $socket->emit('transit_error', ['code' => 'NOT_ENROLLED', 'route_id' => $routeId]);
                return;
            }
        } else {
            socket_log("[TRANSIT_WARN] Redis unavailable — enrollment check skipped (passenger #{$passengerId} route #{$routeId})");
        }

        $socket->join('transit_route_' . $routeId);
        socket_log("[TRANSIT] Passenger #$passengerId subscribed to route #$routeId");
    });

    // 🚌 إلغاء الاشتراك عند إغلاق الراكب للخط
    $socket->on('unsubscribe_transit_route', function ($data) use ($socket, $passengerId) {
        $data    = (array) $data;
        $routeId = (int)($data['route_id'] ?? 0);
        if ($routeId <= 0) return;

        $socket->leave('transit_route_' . $routeId);
        socket_log("[TRANSIT] Passenger #$passengerId unsubscribed from route #$routeId");
    });

    $socket->on('disconnect', function () use ($passengerId, $clientIp) {
        socket_log("[SOCKET_DISCONNECTED] Passenger Disconnected: #$passengerId (IP: $clientIp)");
    });
});

Worker::runAll();