feat: Real-time Workerman gateway + Multi-level exam hierarchy (lesson, unit, final, AI adaptive) and student mastery tracking
This commit is contained in:
Hamza-Ayed
@@ -0,0 +1,271 @@
|
||||
<?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
|
||||
*/
|
||||
|
||||
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') ?: 4040);
|
||||
$internalPort = (int)(getenv('WS_INTERNAL_PORT') ?: 4041);
|
||||
|
||||
$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::decodeJwt($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();
|
||||
Reference in New Issue
Block a user