Hamza/nabeh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement multi-line session failover with round-robin strategy and session readiness validation for OTP delivery.

Browse Source
This commit is contained in:
Hamza-Ayed
2026-08-29 22:11:10 +03:00
parent f9609b4403
commit aec9fcee72
4 changed files with 154 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/Controllers/OTPController.php
+79 -46
View File
@@ -144,38 +144,33 @@ class OTPController extends BaseController
// Clean phone number (remove non-digits including +)
$phone = preg_replace('/\D/', '', $phone);
// 1. Resolve WhatsApp Session
$session = null;
// 1. Resolve WhatsApp Candidate Sessions with Round-Robin & Multi-Line Failover
$candidateSessions = [];
if ($sessionId) {
$session = WhatsAppSession::findSecure((int)$sessionId);
if (!$session || (int)$session['company_id'] !== (int)$companyId) {
$singleSession = WhatsAppSession::findSecure((int)$sessionId);
if (!$singleSession || (int)$singleSession['company_id'] !== (int)$companyId) {
$response->status(404)->json(['error' => 'WhatsApp session not found']);
return;
}
$candidateSessions = [$singleSession];
} else {
// Grab the first connected session of the company
$sessions = WhatsAppSession::findAllByCompany($companyId);
foreach ($sessions as $s) {
if ($s['status'] === 'connected') {
$session = $s;
break;
// Get all active connected sessions ordered by Round-Robin strategy
$candidateSessions = WhatsAppSession::getRoundRobinSessionsList($companyId);
// Fallback: If no connected session found in cache/DB, check all company sessions
if (empty($candidateSessions)) {
$all = WhatsAppSession::findAllByCompany($companyId);
if (!empty($all)) {
$candidateSessions = $all;
}
}
if (!$session && !empty($sessions)) {
$session = $sessions[0]; // fallback to first session if none is connected
}
}
if (!$session) {
if (empty($candidateSessions)) {
$response->status(400)->json(['error' => 'No active WhatsApp sessions configured for this company.']);
return;
}
if ($session['status'] !== 'connected') {
$response->status(400)->json(['error' => 'WhatsApp session is not connected. Connect the session first.']);
return;
}
// 2. Check SaaS subscription quotas
$useElevenLabs = false;
if ($companyId !== 1) {
@@ -208,19 +203,24 @@ class OTPController extends BaseController
// 3. Generate verification code
$code = $customCode ? trim($customCode) : (string)rand(1000, 9999);
// 4. Send Message
// 4. Prepare Payload
try {
$usedElevenLabs = false;
$usedElevenLabsFinal = false;
$audioBase64 = null;
$mimeType = null;
$imageBase64 = null;
$textMsg = null;
if ($type === 'voice') {
// Spacing the digits to force slow Arabic pronunciation: e.g. "1 2 3 4"
$spacedCode = implode(' ', str_split($code));
$textToRead = "رمز التحقق الخاص بك هو: {$spacedCode}. أكرر، رمز التحقق هو: {$spacedCode}.";
$audioBase64 = null;
$mimeType = 'audio/mp3';
if ($useElevenLabs) {
$rule = \App\Models\ChatbotRule::findActiveForRule($companyId, $session['id']);
// Try getting ElevenLabs key from first candidate session
$firstSession = $candidateSessions[0];
$rule = \App\Models\ChatbotRule::findActiveForRule($companyId, $firstSession['id']);
$configuredElKey = ($rule && !empty($rule['elevenlabs_api_key'])) ? $rule['elevenlabs_api_key'] : null;
$elApiKey = \App\Services\GeminiService::getElevenLabsApiKey($configuredElKey);
@@ -233,7 +233,7 @@ class OTPController extends BaseController
if ($audioData) {
$audioBase64 = $audioData['audio'];
$mimeType = $audioData['mimeType'];
$usedElevenLabs = true;
$usedElevenLabsFinal = true;
}
}
}
@@ -242,47 +242,75 @@ class OTPController extends BaseController
// Fallback to Google Translate TTS
$audioBase64 = TTSService::textToSpeechArabic($textToRead);
$mimeType = 'audio/mp3';
$usedElevenLabs = false;
$usedElevenLabsFinal = false;
}
if (!$audioBase64) {
$response->status(500)->json(['error' => 'Failed to generate voice OTP audio.']);
return;
}
// Send voice note
$success = ConversationFlowEngine::sendReply($session, $phone, '', null, $audioBase64, $mimeType);
if (!$success) {
$response->status(500)->json(['error' => 'Failed to send voice OTP via gateway.']);
return;
}
} else if ($type === 'image') {
// Generate OTP Image as Base64
$imageBase64 = $this->generateOtpImage($code);
$success = ConversationFlowEngine::sendReply($session, $phone, '', null, null, null, $imageBase64);
if (!$success) {
$response->status(500)->json(['error' => 'Failed to send image OTP via gateway.']);
return;
}
} else {
// Send text
$customText = $body['message'] ?? null;
if ($customText) {
$textMsg = str_replace('{code}', $code, $customText);
} else {
$textMsg = "رمز التحقق الخاص بك لمتجر نابه هو: *{$code}* \n الرجاء عدم مشاركته مع أي شخص.";
}
$success = ConversationFlowEngine::sendReply($session, $phone, $textMsg);
if (!$success) {
$response->status(500)->json(['error' => 'Failed to send text OTP via gateway.']);
return;
}
// 5. Send Message with Automatic Multi-Line Failover across Active Lines
$sentSuccessfully = false;
$usedSession = null;
$attemptErrors = [];
foreach ($candidateSessions as $index => $session) {
// If there are multiple sessions, skip non-connected ones
if ($session['status'] !== 'connected' && count($candidateSessions) > 1) {
continue;
}
$sessionLabel = ($session['name'] ?? 'Line') . " (" . ($session['session_key'] ?? 'unknown') . ")";
try {
$success = false;
if ($type === 'voice') {
$success = ConversationFlowEngine::sendReply($session, $phone, '', null, $audioBase64, $mimeType);
} else if ($type === 'image') {
$success = ConversationFlowEngine::sendReply($session, $phone, '', null, null, null, $imageBase64);
} else {
$success = ConversationFlowEngine::sendReply($session, $phone, $textMsg);
}
if ($success) {
$sentSuccessfully = true;
$usedSession = $session;
break; // Sent successfully! Break out of failover loop
} else {
$errorDetail = "Failed on {$sessionLabel}";
$attemptErrors[] = $errorDetail;
error_log("[OTP Failover] " . $errorDetail . " → Trying next active line...");
}
} catch (\Exception $sendEx) {
$errorDetail = "Exception on {$sessionLabel}: " . $sendEx->getMessage();
$attemptErrors[] = $errorDetail;
error_log("[OTP Failover Exception] " . $errorDetail);
}
}
if (!$sentSuccessfully) {
$errorSummary = !empty($attemptErrors) ? implode(' | ', $attemptErrors) : 'All candidate lines failed';
$response->status(500)->json([
'error' => "Failed to send {$type} OTP via gateway. Tried " . count($attemptErrors) . " active line(s): {$errorSummary}"
]);
return;
}
// Increment usage stats
if ($companyId !== 1) {
CompanySubscriptionUsage::incrementUsage($companyId, 'request');
if ($type === 'voice' && $usedElevenLabs) {
if ($type === 'voice' && $usedElevenLabsFinal) {
CompanySubscriptionUsage::incrementUsage($companyId, 'voice');
}
}
@@ -291,7 +319,12 @@ class OTPController extends BaseController
'status' => 'success',
'message' => 'OTP sent successfully',
'code' => $code,
'type' => $type
'type' => $type,
'session_used' => [
'id' => $usedSession['id'] ?? null,
'name' => $usedSession['name'] ?? null,
'session_key' => $usedSession['session_key'] ?? null
]
]);
} catch (\Exception $e) {
backend/app/Models/WhatsAppSession.php
+56 -13
View File
@@ -49,26 +49,25 @@ class WhatsAppSession extends BaseModel
}
/**
* Get next connected session using Round-Robin strategy via Redis
* Get and increment Round-Robin index with Redis and file-based fallback
*/
public static function getRoundRobinSession(int $companyId): ?array
public static function getNextRoundRobinIndex(int $companyId, int $count): int
{
$connectedSessions = static::getConnectedSessionsForCompany($companyId);
if (empty($connectedSessions)) {
// Fallback to legacy single session if no connected sessions are found
return static::findByCompany($companyId, false);
}
$count = count($connectedSessions);
if ($count === 1) {
return $connectedSessions[0];
if ($count <= 1) {
return 0;
}
$rrKey = "rr_index:company_{$companyId}";
$currentIndex = Cache::get($rrKey);
if ($currentIndex === null || !is_numeric($currentIndex)) {
// File-based persistent fallback if Redis is unavailable
$tempFile = sys_get_temp_dir() . "/nabeh_rr_company_{$companyId}.txt";
if (file_exists($tempFile)) {
$currentIndex = (int)@file_get_contents($tempFile);
} else {
$currentIndex = 0;
}
} else {
$currentIndex = (int)$currentIndex;
}
@@ -78,7 +77,51 @@ class WhatsAppSession extends BaseModel
Cache::set($rrKey, $nextIndex, 86400);
return $connectedSessions[$selectedIndex];
$tempFile = sys_get_temp_dir() . "/nabeh_rr_company_{$companyId}.txt";
@file_put_contents($tempFile, (string)$nextIndex, LOCK_EX);
return $selectedIndex;
}
/**
* Get all active connected sessions ordered by Round-Robin index for seamless failover
*/
public static function getRoundRobinSessionsList(int $companyId): array
{
$connectedSessions = static::getConnectedSessionsForCompany($companyId);
$count = count($connectedSessions);
if ($count === 0) {
return [];
}
if ($count === 1) {
return $connectedSessions;
}
$startIndex = static::getNextRoundRobinIndex($companyId, $count);
// Reorder list starting from $startIndex: [start..end, 0..start-1]
$ordered = [];
for ($i = 0; $i < $count; $i++) {
$ordered[] = $connectedSessions[($startIndex + $i) % $count];
}
return $ordered;
}
/**
* Get next connected session using Round-Robin strategy via Redis
*/
public static function getRoundRobinSession(int $companyId): ?array
{
$list = static::getRoundRobinSessionsList($companyId);
if (!empty($list)) {
return $list[0];
}
// Fallback to legacy single session if no connected sessions are found
return static::findByCompany($companyId, false);
}
/**
whatsapp-gateway/baileys-client.js
+6
View File
@@ -804,11 +804,17 @@ function getActiveSessions() {
return Array.from(sessions.keys());
}
function isSessionReady(session_key) {
const sock = sessions.get(session_key);
return !!(sock && sock.user && sock.user.id);
}
module.exports = {
startSession,
disconnectSession,
sendMessage,
getActiveSessions,
isSessionReady,
checkContact,
exportChatHistory,
throttleManager
whatsapp-gateway/server.js
+12 -2
View File
@@ -20,7 +20,7 @@ for (const p of envPaths) {
const express = require('express');
const cors = require('cors');
const { startSession, disconnectSession, sendMessage, getActiveSessions, checkContact, exportChatHistory, throttleManager } = require('./baileys-client');
const { startSession, disconnectSession, sendMessage, getActiveSessions, isSessionReady, checkContact, exportChatHistory, throttleManager } = require('./baileys-client');
const app = express();
app.use(cors());
@@ -79,7 +79,17 @@ app.post('/api/sessions/disconnect', async (req, res) => {
// Get list of active session keys in memory
app.get('/api/sessions/active', (req, res) => {
res.json({ status: 'success', active_sessions: getActiveSessions() });
const active = getActiveSessions();
const readySessions = active.filter(key => isSessionReady(key));
res.json({
status: 'success',
active_sessions: active,
ready_sessions: readySessions,
details: active.map(key => ({
session_key: key,
ready: isSessionReady(key)
}))
});
});
// Get throttle queue status for all sessions (Anti-Ban monitoring)