@@ -1,113 +0,0 @@
|
|||||||
<?php
|
|
||||||
/**
|
|
||||||
* SSE endpoint: stream events for a specific user's conversation
|
|
||||||
* - Usage: /api/stream_user_messages.php?user_id=3180&since=YYYY-MM-DD%20HH:MM:SS
|
|
||||||
* - Security: requires session authentication (requireAuthentication)
|
|
||||||
* - Behavior: tries to use Redis pub/sub if available, otherwise polls DB every 2s for new messages.
|
|
||||||
*/
|
|
||||||
require_once __DIR__ . '/../config/config.php';
|
|
||||||
requireAuthentication();
|
|
||||||
|
|
||||||
header('Content-Type: text/event-stream');
|
|
||||||
header('Cache-Control: no-cache');
|
|
||||||
header('Connection: keep-alive');
|
|
||||||
// Tell proxies not to buffer
|
|
||||||
if (function_exists('apache_setenv')) @apache_setenv('no-gzip', '1');
|
|
||||||
|
|
||||||
try {
|
|
||||||
$userId = intval($_GET['user_id'] ?? 0);
|
|
||||||
if (!$userId) {
|
|
||||||
http_response_code(400);
|
|
||||||
echo "data: {\"error\": \"user_id es requerido\"}\n\n";
|
|
||||||
exit;
|
|
||||||
}
|
|
||||||
|
|
||||||
ignore_user_abort(true);
|
|
||||||
set_time_limit(0);
|
|
||||||
|
|
||||||
$db = Database::getInstance();
|
|
||||||
|
|
||||||
// initialize last_ts: if client passed since, use it; else use latest message timestamp or now-1s
|
|
||||||
$last = !empty($_GET['since']) ? $_GET['since'] : null;
|
|
||||||
if (is_null($last)) {
|
|
||||||
try {
|
|
||||||
// Database wrapper exposes fetch() for a single row
|
|
||||||
$r = $db->fetch('SELECT created_at FROM conversations WHERE user_id = :uid ORDER BY created_at DESC LIMIT 1', ['uid' => $userId]);
|
|
||||||
$last = $r && isset($r['created_at']) ? $r['created_at'] : date('Y-m-d H:i:s', time() - 1);
|
|
||||||
} catch (Exception $e) {
|
|
||||||
$last = date('Y-m-d H:i:s', time() - 1);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Try Redis pub/sub first (if available)
|
|
||||||
$usedRedis = false;
|
|
||||||
if (extension_loaded('redis')) {
|
|
||||||
try {
|
|
||||||
$redis = new Redis();
|
|
||||||
// connect using defaults; expect redis on localhost:6379
|
|
||||||
$redis->connect('127.0.0.1');
|
|
||||||
// subscribe in a non-blocking manner via pub/sub loop
|
|
||||||
$chan = "user_{$userId}";
|
|
||||||
// Use raw subscribe loop with callback
|
|
||||||
$usedRedis = true;
|
|
||||||
$redis->subscribe([$chan], function ($redis, $chanName, $msg) use (&$last) {
|
|
||||||
// msg expected JSON string with {type:'new_message', user_id, message_id, created_at}
|
|
||||||
$payload = $msg;
|
|
||||||
if (is_array($msg)) $payload = json_encode($msg);
|
|
||||||
echo "event: new_message\n";
|
|
||||||
echo "data: {$payload}\n\n";
|
|
||||||
@ob_flush(); @flush();
|
|
||||||
});
|
|
||||||
} catch (Exception $e) {
|
|
||||||
error_log('stream_user_messages.php - redis subscribe failed: ' . $e->getMessage());
|
|
||||||
$usedRedis = false;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Fallback: polling loop (also used if redis is not available or subscription ended)
|
|
||||||
$start = time();
|
|
||||||
$timeout = 55; // keep connection alive for ~55s (EventSource will reconnect)
|
|
||||||
while (!connection_aborted()) {
|
|
||||||
// Query for messages strictly newer than last timestamp
|
|
||||||
try {
|
|
||||||
$rows = $db->fetchAll(
|
|
||||||
'SELECT id, created_at FROM conversations WHERE user_id = :uid AND created_at > :since ORDER BY created_at ASC',
|
|
||||||
['uid' => $userId, 'since' => $last]
|
|
||||||
);
|
|
||||||
} catch (Exception $e) {
|
|
||||||
error_log('stream_user_messages.php - DB poll failed: ' . $e->getMessage());
|
|
||||||
$rows = [];
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!empty($rows)) {
|
|
||||||
foreach ($rows as $r) {
|
|
||||||
$d = json_encode(['type' => 'new_message', 'user_id' => $userId, 'message_id' => (int)$r['id'], 'created_at' => $r['created_at']]);
|
|
||||||
echo "event: new_message\n";
|
|
||||||
echo "data: {$d}\n\n";
|
|
||||||
@ob_flush(); @flush();
|
|
||||||
$last = $r['created_at'];
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Periodic ping to keep connection alive
|
|
||||||
echo "event: ping\n";
|
|
||||||
echo "data: {\"now\": \"" . date('Y-m-d H:i:s') . "\"}\n\n";
|
|
||||||
@ob_flush(); @flush();
|
|
||||||
|
|
||||||
// Break/refresh connection after timeout so client reconnects (avoids very long running PHP requests)
|
|
||||||
if ((time() - $start) > $timeout) break;
|
|
||||||
|
|
||||||
sleep(2);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Final flush
|
|
||||||
echo "event: close\n";
|
|
||||||
echo "data: {}\n\n";
|
|
||||||
@ob_flush(); @flush();
|
|
||||||
exit;
|
|
||||||
} catch (Exception $e) {
|
|
||||||
error_log('stream_user_messages.php - fatal: ' . $e->getMessage());
|
|
||||||
http_response_code(500);
|
|
||||||
echo "data: {\"error\": \"Internal server error\"}\n\n";
|
|
||||||
exit;
|
|
||||||
}
|
|
||||||
@@ -305,18 +305,6 @@ class WhatsAppWebhook {
|
|||||||
error_log('Failed to create notification: ' . $e->getMessage());
|
error_log('Failed to create notification: ' . $e->getMessage());
|
||||||
}
|
}
|
||||||
|
|
||||||
// Try to publish a lightweight event to Redis so SSE/WebSocket listeners can be notified immediately
|
|
||||||
try {
|
|
||||||
if (extension_loaded('redis')) {
|
|
||||||
$payload = json_encode(['type' => 'new_message', 'user_id' => $user['id'], 'message_id' => (int)$conversationId, 'created_at' => date('Y-m-d H:i:s')]);
|
|
||||||
$r = new Redis();
|
|
||||||
$r->connect('127.0.0.1');
|
|
||||||
$r->publish('user_' . $user['id'], $payload);
|
|
||||||
}
|
|
||||||
} catch (Exception $e) {
|
|
||||||
error_log('[webhook] redis publish failed: ' . $e->getMessage());
|
|
||||||
}
|
|
||||||
|
|
||||||
// Procesar con bot (protección contra excepciones externas)
|
// Procesar con bot (protección contra excepciones externas)
|
||||||
try {
|
try {
|
||||||
$this->botService->processMessage($user, $messageText, $messageType);
|
$this->botService->processMessage($user, $messageText, $messageType);
|
||||||
|
|||||||
@@ -1034,8 +1034,6 @@
|
|||||||
this._stagedMessages = [];
|
this._stagedMessages = [];
|
||||||
// Last message timestamp seen (for delta polling)
|
// Last message timestamp seen (for delta polling)
|
||||||
this._latestMessage = null;
|
this._latestMessage = null;
|
||||||
// SSE connection handle for server push (EventSource)
|
|
||||||
this._sse = null;
|
|
||||||
// Removed: automatic "Nuevos mensajes" indicator — we always reload full conversation now
|
// Removed: automatic "Nuevos mensajes" indicator — we always reload full conversation now
|
||||||
// this._lastRenderMessageCount = 0;
|
// this._lastRenderMessageCount = 0;
|
||||||
|
|
||||||
@@ -1900,50 +1898,6 @@
|
|||||||
}, 10000);
|
}, 10000);
|
||||||
}
|
}
|
||||||
|
|
||||||
// SSE: start listening for server-sent events for the current conversation
|
|
||||||
startSSE(userId) {
|
|
||||||
try {
|
|
||||||
this.stopSSE();
|
|
||||||
if (!userId || typeof EventSource === 'undefined') return;
|
|
||||||
const url = `api/stream_user_messages.php?user_id=${userId}`;
|
|
||||||
try {
|
|
||||||
this._sse = new EventSource(url);
|
|
||||||
this._sse.onmessage = (ev) => {
|
|
||||||
try {
|
|
||||||
const d = JSON.parse(ev.data || '{}');
|
|
||||||
if (d && d.type === 'new_message' && d.user_id == this.currentUserId) {
|
|
||||||
if (console && console.debug) console.debug('SSE new_message received for user', d.user_id, '-> triggering full load');
|
|
||||||
if (!this._fullLoadInProgress) {
|
|
||||||
this.loadMessages(this.currentUserId, true, true).catch(e => console.warn('SSE triggered load failed', e));
|
|
||||||
} else {
|
|
||||||
if (console && console.debug) console.debug('SSE: full load already in progress, skipping immediate load');
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} catch (e) { console.warn('SSE message parse failed', e); }
|
|
||||||
};
|
|
||||||
this._sse.addEventListener('ping', () => {/* no-op */});
|
|
||||||
this._sse.onerror = (err) => {
|
|
||||||
console.warn('SSE error', err);
|
|
||||||
// Close and attempt reconnect after short delay
|
|
||||||
try { this._sse.close(); } catch(e) {}
|
|
||||||
this._sse = null;
|
|
||||||
setTimeout(() => { if (this.currentUserId == userId) this.startSSE(userId); }, 3000);
|
|
||||||
};
|
|
||||||
if (console && console.debug) console.debug('SSE started for user', userId);
|
|
||||||
} catch (e) { console.warn('Failed to start EventSource', e); }
|
|
||||||
} catch (e) { console.warn('startSSE failed', e); }
|
|
||||||
}
|
|
||||||
|
|
||||||
stopSSE() {
|
|
||||||
try {
|
|
||||||
if (this._sse) {
|
|
||||||
try { this._sse.close(); } catch(e) {}
|
|
||||||
this._sse = null;
|
|
||||||
if (console && console.debug) console.debug('SSE stopped');
|
|
||||||
}
|
|
||||||
} catch (e) { /* ignore */ }
|
|
||||||
}
|
|
||||||
|
|
||||||
// New: periodic delta poller to fetch only messages newer than last seen timestamp
|
// New: periodic delta poller to fetch only messages newer than last seen timestamp
|
||||||
async pollNewMessages() {
|
async pollNewMessages() {
|
||||||
if (!this.currentUserId) return;
|
if (!this.currentUserId) return;
|
||||||
@@ -2303,8 +2257,6 @@
|
|||||||
|
|
||||||
async openConversation(userId, userName, phoneNumber) {
|
async openConversation(userId, userName, phoneNumber) {
|
||||||
this.currentUserId = userId;
|
this.currentUserId = userId;
|
||||||
// Stop any previous SSE connection before opening a new conversation
|
|
||||||
try { this.stopSSE(); } catch(e) { /* ignore */ }
|
|
||||||
// Clear any staged messages from a previous conversation and hide indicator
|
// Clear any staged messages from a previous conversation and hide indicator
|
||||||
try { if (this._stagedMessageIds) { this._stagedMessageIds.clear(); this._stagedCount = 0; this.hideNewMessagesIndicator(); } } catch(e) {}
|
try { if (this._stagedMessageIds) { this._stagedMessageIds.clear(); this._stagedCount = 0; this.hideNewMessagesIndicator(); } } catch(e) {}
|
||||||
|
|
||||||
@@ -2640,9 +2592,6 @@
|
|||||||
// Resume auto-refresh after the load
|
// Resume auto-refresh after the load
|
||||||
try { this._suspendAutoRefresh = false; } catch(e) {}
|
try { this._suspendAutoRefresh = false; } catch(e) {}
|
||||||
|
|
||||||
// Start SSE connection for this conversation to receive server push events
|
|
||||||
try { this.startSSE(userId); } catch(e) { console.warn('startSSE failed to initialize', e); }
|
|
||||||
|
|
||||||
// Añadir listener para scroll arriba (cargar más historial)
|
// Añadir listener para scroll arriba (cargar más historial)
|
||||||
const chatContainer = document.getElementById('chat-conversations');
|
const chatContainer = document.getElementById('chat-conversations');
|
||||||
if (chatContainer) {
|
if (chatContainer) {
|
||||||
|
|||||||
Reference in New Issue
Block a user