diff --git a/api/stream_user_messages.php b/api/stream_user_messages.php deleted file mode 100644 index 8c755a0..0000000 --- a/api/stream_user_messages.php +++ /dev/null @@ -1,113 +0,0 @@ -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; -} diff --git a/api/webhook.php b/api/webhook.php index 9efb49e..7984023 100644 --- a/api/webhook.php +++ b/api/webhook.php @@ -304,18 +304,6 @@ class WhatsAppWebhook { } catch (Exception $e) { 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) try { diff --git a/conversations.php b/conversations.php index 5d2d447..4819592 100644 --- a/conversations.php +++ b/conversations.php @@ -1034,8 +1034,6 @@ this._stagedMessages = []; // Last message timestamp seen (for delta polling) this._latestMessage = null; - // SSE connection handle for server push (EventSource) - this._sse = null; // Removed: automatic "Nuevos mensajes" indicator — we always reload full conversation now // this._lastRenderMessageCount = 0; @@ -1900,50 +1898,6 @@ }, 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 async pollNewMessages() { if (!this.currentUserId) return; @@ -2303,8 +2257,6 @@ async openConversation(userId, userName, phoneNumber) { 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 try { if (this._stagedMessageIds) { this._stagedMessageIds.clear(); this._stagedCount = 0; this.hideNewMessagesIndicator(); } } catch(e) {} @@ -2640,9 +2592,6 @@ // Resume auto-refresh after the load 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) const chatContainer = document.getElementById('chat-conversations'); if (chatContainer) {