278 lines
11 KiB
PHP
278 lines
11 KiB
PHP
<?php
|
|
/**
|
|
* Saqel Platform - High Performance Real-Time WebSocket Server (Workerman)
|
|
* =========================================================================
|
|
* Ports:
|
|
* - 4040 : Public WebSocket for Students, Teachers & Guardians
|
|
* - 4041 : Internal HTTP Event Dispatcher for PHP Backend
|
|
*/
|
|
|
|
if (file_exists(__DIR__ . '/../vendor/autoload.php')) {
|
|
require_once __DIR__ . '/../vendor/autoload.php';
|
|
} elseif (file_exists(__DIR__ . '/vendor/autoload.php')) {
|
|
require_once __DIR__ . '/vendor/autoload.php';
|
|
}
|
|
|
|
use Workerman\Worker;
|
|
use Workerman\Connection\TcpConnection;
|
|
|
|
require_once __DIR__ . '/../app/bootstrap.php';
|
|
require_once __DIR__ . '/../app/Core/Security.php';
|
|
require_once __DIR__ . '/../app/Core/Database.php';
|
|
|
|
// 1. Initialize Public WebSocket Server
|
|
$wsPort = (int)(getenv('WS_PORT') ?: 4240);
|
|
$internalPort = (int)(getenv('WS_INTERNAL_PORT') ?: 4241);
|
|
|
|
$wsWorker = new Worker("websocket://0.0.0.0:{$wsPort}");
|
|
$wsWorker->count = 1; // Single process for unified in-memory connection registry
|
|
|
|
// Registry: [userId => [connectionId => TcpConnection]]
|
|
$userConnections = [];
|
|
|
|
// Registry: [connectionId => userId]
|
|
$connectionUserMap = [];
|
|
|
|
$wsWorker->onWorkerStart = function () use ($wsWorker, $internalPort, &$userConnections) {
|
|
echo "====================================================\n";
|
|
echo " 🚀 SAQEL REAL-TIME WORKERMAN SERVER STARTED \n";
|
|
echo " 📡 WebSocket: 0.0.0.0:{$GLOBALS['wsPort']} \n";
|
|
echo " 🔒 Internal HTTP: 0.0.0.0:{$internalPort} \n";
|
|
echo "====================================================\n";
|
|
|
|
// Internal HTTP Event Listener (Receives pushes from PHP backend controllers)
|
|
$innerHttp = new Worker("http://0.0.0.0:{$internalPort}");
|
|
$innerHttp->onMessage = function ($connection, $request) use (&$userConnections) {
|
|
$post = $request->post();
|
|
$action = trim($post['action'] ?? '');
|
|
$targetUserId = (int)($post['target_user_id'] ?? 0);
|
|
$payload = $post['payload'] ?? [];
|
|
|
|
if (is_string($payload)) {
|
|
$payload = json_decode($payload, true) ?: $payload;
|
|
}
|
|
|
|
switch ($action) {
|
|
case 'push_chat':
|
|
if ($targetUserId && isset($userConnections[$targetUserId])) {
|
|
foreach ($userConnections[$targetUserId] as $conn) {
|
|
$conn->send(json_encode([
|
|
'event' => 'chat_message',
|
|
'data' => $payload
|
|
], JSON_UNESCAPED_UNICODE));
|
|
}
|
|
$connection->send(json_encode(['status' => 'delivered', 'recipients' => count($userConnections[$targetUserId])]));
|
|
return;
|
|
}
|
|
$connection->send(json_encode(['status' => 'offline']));
|
|
break;
|
|
|
|
case 'push_notification':
|
|
if ($targetUserId && isset($userConnections[$targetUserId])) {
|
|
foreach ($userConnections[$targetUserId] as $conn) {
|
|
$conn->send(json_encode([
|
|
'event' => 'notification',
|
|
'data' => $payload
|
|
], JSON_UNESCAPED_UNICODE));
|
|
}
|
|
$connection->send(json_encode(['status' => 'delivered']));
|
|
return;
|
|
}
|
|
$connection->send(json_encode(['status' => 'offline']));
|
|
break;
|
|
|
|
case 'drm_force_logout':
|
|
if ($targetUserId && isset($userConnections[$targetUserId])) {
|
|
foreach ($userConnections[$targetUserId] as $conn) {
|
|
$conn->send(json_encode([
|
|
'event' => 'drm_session_terminated',
|
|
'message' => 'تم تسجيل الدخول من جهاز آخر. تم إنهاء الجلسة لحماية محتواك.'
|
|
], JSON_UNESCAPED_UNICODE));
|
|
}
|
|
$connection->send(json_encode(['status' => 'kicked']));
|
|
return;
|
|
}
|
|
$connection->send(json_encode(['status' => 'not_found']));
|
|
break;
|
|
|
|
case 'broadcast_course':
|
|
$courseId = (int)($post['course_id'] ?? 0);
|
|
// Broadcast to all active connections
|
|
foreach ($userConnections as $uid => $conns) {
|
|
foreach ($conns as $conn) {
|
|
$conn->send(json_encode([
|
|
'event' => 'course_broadcast',
|
|
'course_id' => $courseId,
|
|
'data' => $payload
|
|
], JSON_UNESCAPED_UNICODE));
|
|
}
|
|
}
|
|
$connection->send(json_encode(['status' => 'broadcasted']));
|
|
break;
|
|
|
|
default:
|
|
$connection->send(json_encode(['status' => 'unknown_action']));
|
|
break;
|
|
}
|
|
};
|
|
$innerHttp->listen();
|
|
};
|
|
|
|
$wsWorker->onConnect = function (TcpConnection $connection) {
|
|
// Initial ping on connection
|
|
$connection->send(json_encode([
|
|
'event' => 'connected',
|
|
'message' => 'Connected to Saqel Real-time Network. Please authenticate with your JWT token.'
|
|
], JSON_UNESCAPED_UNICODE));
|
|
};
|
|
|
|
$wsWorker->onMessage = function (TcpConnection $connection, $data) use (&$userConnections, &$connectionUserMap) {
|
|
$msg = json_decode($data, true);
|
|
if (!is_array($msg)) {
|
|
return;
|
|
}
|
|
|
|
$event = $msg['event'] ?? '';
|
|
$payload = $msg['data'] ?? [];
|
|
|
|
switch ($event) {
|
|
// 1. Authenticate WebSocket Connection via JWT
|
|
case 'auth':
|
|
$token = trim((string)($payload['token'] ?? ''));
|
|
if (empty($token)) {
|
|
$connection->send(json_encode(['event' => 'auth_error', 'message' => 'Token required']));
|
|
return;
|
|
}
|
|
|
|
try {
|
|
$decoded = \App\Core\Security::verifyJWT($token);
|
|
if (!$decoded || empty($decoded['user_id'])) {
|
|
$connection->send(json_encode(['event' => 'auth_error', 'message' => 'Invalid or expired token']));
|
|
return;
|
|
}
|
|
|
|
$userId = (int)$decoded['user_id'];
|
|
$role = (string)($decoded['role'] ?? 'student');
|
|
|
|
$userConnections[$userId][$connection->id] = $connection;
|
|
$connectionUserMap[$connection->id] = $userId;
|
|
|
|
$connection->send(json_encode([
|
|
'event' => 'authenticated',
|
|
'user_id' => $userId,
|
|
'role' => $role,
|
|
'message' => 'Authenticated successfully. Real-time stream active.'
|
|
], JSON_UNESCAPED_UNICODE));
|
|
|
|
echo "✅ [WS Auth] User #{$userId} ({$role}) connected (Connection ID: {$connection->id})\n";
|
|
|
|
} catch (\Exception $e) {
|
|
$connection->send(json_encode(['event' => 'auth_error', 'message' => $e->getMessage()]));
|
|
}
|
|
break;
|
|
|
|
// 2. Direct Chat Message (Student <-> Teacher)
|
|
case 'chat_send':
|
|
$senderId = $connectionUserMap[$connection->id] ?? null;
|
|
if (!$senderId) {
|
|
$connection->send(json_encode(['event' => 'error', 'message' => 'Unauthorized']));
|
|
return;
|
|
}
|
|
|
|
$receiverId = (int)($payload['receiver_id'] ?? 0);
|
|
$messageText = trim((string)($payload['message'] ?? ''));
|
|
$messageType = (string)($payload['message_type'] ?? 'text');
|
|
$mediaUrl = !empty($payload['media_url']) ? trim((string)$payload['media_url']) : null;
|
|
$courseId = !empty($payload['course_id']) ? (int)$payload['course_id'] : null;
|
|
|
|
if (!$receiverId || empty($messageText)) {
|
|
$connection->send(json_encode(['event' => 'error', 'message' => 'receiver_id and message required']));
|
|
return;
|
|
}
|
|
|
|
// Save to MySQL
|
|
$uuid = sprintf('%04x%04x-%04x-%04x-%04x-%04x%04x%04x',
|
|
mt_rand(0, 0xffff), mt_rand(0, 0xffff),
|
|
mt_rand(0, 0xffff),
|
|
mt_rand(0, 0x0fff) | 0x4000,
|
|
mt_rand(0, 0x3fff) | 0x8000,
|
|
mt_rand(0, 0xffff), mt_rand(0, 0xffff), mt_rand(0, 0xffff)
|
|
);
|
|
|
|
try {
|
|
$msgId = \App\Core\Database::insert(
|
|
"INSERT INTO chat_messages (uuid, sender_id, receiver_id, course_id, message, message_type, media_url, is_read)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, 0)",
|
|
[$uuid, $senderId, $receiverId, $courseId, $messageText, $messageType, $mediaUrl]
|
|
);
|
|
|
|
$msgData = [
|
|
'id' => $msgId,
|
|
'uuid' => $uuid,
|
|
'sender_id' => $senderId,
|
|
'receiver_id' => $receiverId,
|
|
'course_id' => $courseId,
|
|
'message' => $messageText,
|
|
'message_type' => $messageType,
|
|
'media_url' => $mediaUrl,
|
|
'is_read' => false,
|
|
'created_at' => date('Y-m-d H:i:s')
|
|
];
|
|
|
|
// Echo back to sender with confirmed ID
|
|
$connection->send(json_encode([
|
|
'event' => 'chat_sent',
|
|
'data' => array_merge($msgData, ['is_mine' => true])
|
|
], JSON_UNESCAPED_UNICODE));
|
|
|
|
// Dispatch to receiver if online
|
|
if (isset($userConnections[$receiverId])) {
|
|
foreach ($userConnections[$receiverId] as $recConn) {
|
|
$recConn->send(json_encode([
|
|
'event' => 'chat_message',
|
|
'data' => array_merge($msgData, ['is_mine' => false])
|
|
], JSON_UNESCAPED_UNICODE));
|
|
}
|
|
}
|
|
|
|
} catch (\Exception $e) {
|
|
$connection->send(json_encode(['event' => 'error', 'message' => 'DB insert failed: ' . $e->getMessage()]));
|
|
}
|
|
break;
|
|
|
|
// 3. Typing Indicator
|
|
case 'typing':
|
|
$senderId = $connectionUserMap[$connection->id] ?? null;
|
|
$receiverId = (int)($payload['receiver_id'] ?? 0);
|
|
if ($senderId && $receiverId && isset($userConnections[$receiverId])) {
|
|
foreach ($userConnections[$receiverId] as $recConn) {
|
|
$recConn->send(json_encode([
|
|
'event' => 'user_typing',
|
|
'sender_id' => $senderId,
|
|
'is_typing' => (bool)($payload['is_typing'] ?? true)
|
|
]));
|
|
}
|
|
}
|
|
break;
|
|
|
|
// 4. Heartbeat
|
|
case 'ping':
|
|
$connection->send(json_encode(['event' => 'pong', 'timestamp' => time()]));
|
|
break;
|
|
}
|
|
};
|
|
|
|
$wsWorker->onClose = function (TcpConnection $connection) use (&$userConnections, &$connectionUserMap) {
|
|
if (isset($connectionUserMap[$connection->id])) {
|
|
$userId = $connectionUserMap[$connection->id];
|
|
unset($userConnections[$userId][$connection->id]);
|
|
if (empty($userConnections[$userId])) {
|
|
unset($userConnections[$userId]);
|
|
}
|
|
unset($connectionUserMap[$connection->id]);
|
|
echo "🔌 [WS Disconnect] User #{$userId} disconnected (Connection ID: {$connection->id})\n";
|
|
}
|
|
};
|
|
|
|
Worker::runAll();
|