Реализован SSE + Redis Pub/Sub для AI Drawer
- Добавлен SSE endpoint (aiassist/ai_sse.php) для real-time получения ответов от n8n - Обновлен n8n_proxy.php: убран callback, добавлена передача Redis параметров в n8n - Обновлен ai-drawer-simple.js: переход с polling на SSE с fallback через Redis - Добавлен check_redis_response.php для прямого чтения из Redis кэша - Добавлена документация: N8N_REDIS_SETUP.md, N8N_REDIS_FIX.md, AI_DRAWER_REDIS_SSE.md - Поддержка plain text ответов от n8n (автоматическое определение формата) - Кэширование ответов в Redis для надежности (TTL 5 минут)
This commit is contained in:
Fedor
@@ -4,6 +4,7 @@ class AIDrawer {
|
||||
this.fontSize = 'normal';
|
||||
this.avatarType = 'default';
|
||||
this.sessionId = null;
|
||||
this.currentEventSource = null; // Для SSE соединения
|
||||
this.init();
|
||||
|
||||
// Загружаем историю сразу при инициализации (при загрузке страницы)
|
||||
@@ -425,9 +426,9 @@ class AIDrawer {
|
||||
console.log('AI Drawer: data.success =', data.success, 'type:', typeof data.success);
|
||||
|
||||
if (data.success && data.task_id) {
|
||||
// Запрос принят, начинаем polling по task_id
|
||||
// Запрос принят, подписываемся на SSE события через Redis
|
||||
console.log('AI Drawer: Request accepted, task_id:', data.task_id);
|
||||
this.startPolling(data.task_id);
|
||||
this.startSSEListener(data.task_id);
|
||||
} else {
|
||||
throw new Error(data.message || 'Unknown error');
|
||||
}
|
||||
@@ -440,18 +441,189 @@ class AIDrawer {
|
||||
}
|
||||
}
|
||||
|
||||
// Метод для polling результатов
|
||||
async startPolling(taskId) {
|
||||
console.log('AI Drawer: Starting polling for task:', taskId);
|
||||
// Метод для подписки на SSE события через Redis Pub/Sub
|
||||
startSSEListener(taskId) {
|
||||
console.log('AI Drawer: Starting SSE listener for task:', taskId);
|
||||
|
||||
// Закрываем предыдущее соединение если есть
|
||||
if (this.currentEventSource) {
|
||||
this.currentEventSource.close();
|
||||
}
|
||||
|
||||
// Флаг для отслеживания получения ответа
|
||||
let responseReceived = false;
|
||||
|
||||
// Создаем новое SSE соединение
|
||||
const sseUrl = `/aiassist/ai_sse.php?task_id=${encodeURIComponent(taskId)}`;
|
||||
this.currentEventSource = new EventSource(sseUrl);
|
||||
|
||||
// Обработчик подключения
|
||||
this.currentEventSource.onopen = () => {
|
||||
console.log('AI Drawer: SSE connection opened');
|
||||
};
|
||||
|
||||
// Обработчик получения ответа
|
||||
this.currentEventSource.addEventListener('response', (event) => {
|
||||
try {
|
||||
const data = JSON.parse(event.data);
|
||||
console.log('AI Drawer: Received response via SSE:', data);
|
||||
|
||||
if (data.data && data.data.response) {
|
||||
responseReceived = true; // Отмечаем что получили ответ
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage(data.data.response, false, 25);
|
||||
this.currentEventSource.close();
|
||||
this.currentEventSource = null;
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('AI Drawer: SSE response parse error:', error);
|
||||
}
|
||||
});
|
||||
|
||||
// Обработчик ошибок SSE (стандартное событие)
|
||||
this.currentEventSource.onerror = (event) => {
|
||||
console.error('AI Drawer: SSE connection error:', event);
|
||||
console.log('AI Drawer: SSE readyState:', this.currentEventSource?.readyState);
|
||||
|
||||
// НЕ вызываем fallback если уже получили ответ (SSE закрывается после отправки)
|
||||
if (responseReceived) {
|
||||
console.log('AI Drawer: Response already received, ignoring SSE close error');
|
||||
return;
|
||||
}
|
||||
|
||||
// Fallback на polling только если SSE действительно не работает
|
||||
if (this.currentEventSource && this.currentEventSource.readyState === EventSource.CLOSED) {
|
||||
console.log('AI Drawer: SSE closed without response, falling back to Redis check');
|
||||
this.currentEventSource.close();
|
||||
this.currentEventSource = null;
|
||||
// Вместо polling БД проверяем Redis напрямую
|
||||
this.checkRedisDirectly(taskId);
|
||||
}
|
||||
};
|
||||
|
||||
// Обработчик кастомных ошибок от сервера
|
||||
this.currentEventSource.addEventListener('error', (event) => {
|
||||
try {
|
||||
const data = JSON.parse(event.data);
|
||||
console.error('AI Drawer: SSE error event from server:', data);
|
||||
|
||||
if (data.data && data.data.error) {
|
||||
responseReceived = true; // Отмечаем что получили ответ (даже если ошибка)
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage(data.data.error || 'Произошла ошибка при обработке запроса.', false, 25);
|
||||
this.currentEventSource.close();
|
||||
this.currentEventSource = null;
|
||||
}
|
||||
} catch (error) {
|
||||
// Игнорируем ошибки парсинга
|
||||
}
|
||||
});
|
||||
|
||||
// Обработчик heartbeat (поддержание соединения)
|
||||
this.currentEventSource.addEventListener('heartbeat', (event) => {
|
||||
console.log('AI Drawer: SSE heartbeat received');
|
||||
});
|
||||
|
||||
// Обработчик подключения
|
||||
this.currentEventSource.addEventListener('connected', (event) => {
|
||||
console.log('AI Drawer: SSE connected:', event.data);
|
||||
});
|
||||
|
||||
// Таймаут на случай если SSE не работает (fallback на Redis check)
|
||||
setTimeout(() => {
|
||||
if (this.currentEventSource && this.currentEventSource.readyState === EventSource.CONNECTING && !responseReceived) {
|
||||
console.log('AI Drawer: SSE timeout, checking Redis directly');
|
||||
this.currentEventSource.close();
|
||||
this.currentEventSource = null;
|
||||
this.checkRedisDirectly(taskId);
|
||||
}
|
||||
}, 5000); // 5 секунд на подключение
|
||||
|
||||
// Общий таймаут ожидания ответа (5 минут)
|
||||
setTimeout(() => {
|
||||
if (this.currentEventSource && !responseReceived) {
|
||||
this.currentEventSource.close();
|
||||
this.currentEventSource = null;
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage('Время ожидания истекло. Попробуйте еще раз.', false, 25);
|
||||
}
|
||||
}, 300000);
|
||||
}
|
||||
|
||||
// Метод для прямой проверки Redis (если SSE не работает)
|
||||
async checkRedisDirectly(taskId) {
|
||||
console.log('AI Drawer: Checking Redis directly for task:', taskId);
|
||||
|
||||
// Проверяем несколько раз с интервалом (на случай если ответ еще обрабатывается)
|
||||
let attempts = 0;
|
||||
const maxAttempts = 30; // 30 попыток = 1 минута (каждые 2 секунды)
|
||||
|
||||
const checkInterval = setInterval(async () => {
|
||||
attempts++;
|
||||
console.log(`AI Drawer: Redis check attempt ${attempts}/${maxAttempts}`);
|
||||
|
||||
try {
|
||||
const response = await fetch(`/aiassist/check_redis_response.php?task_id=${encodeURIComponent(taskId)}`);
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Redis check failed: ${response.status}`);
|
||||
}
|
||||
|
||||
const data = await response.json();
|
||||
|
||||
if (data.found && data.response) {
|
||||
clearInterval(checkInterval);
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage(data.response, false, 25);
|
||||
} else if (attempts >= maxAttempts) {
|
||||
clearInterval(checkInterval);
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage('Ответ не получен. Попробуйте отправить запрос еще раз.', false, 25);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('AI Drawer: Redis direct check error:', error);
|
||||
if (attempts >= maxAttempts) {
|
||||
clearInterval(checkInterval);
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage('Ошибка при получении ответа. Попробуйте еще раз.', false, 25);
|
||||
}
|
||||
}
|
||||
}, 2000); // Проверяем каждые 2 секунды
|
||||
}
|
||||
|
||||
// Fallback метод для polling (если SSE не работает)
|
||||
async startPollingFallback(taskId) {
|
||||
console.log('AI Drawer: Starting polling fallback for task:', taskId);
|
||||
|
||||
const pollInterval = setInterval(async () => {
|
||||
try {
|
||||
const completed = await this.checkAIResult(taskId);
|
||||
if (completed) {
|
||||
const response = await fetch(`/get_ai_result.php?task_id=${taskId}`);
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Result check failed: ${response.status}`);
|
||||
}
|
||||
|
||||
const data = await response.json();
|
||||
|
||||
if (data.status === 'completed') {
|
||||
clearInterval(pollInterval);
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage(data.response, false, 25);
|
||||
} else if (data.status === 'error') {
|
||||
clearInterval(pollInterval);
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage(data.error || 'Произошла ошибка при обработке запроса.', false, 25);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('AI Drawer: Polling error:', error);
|
||||
console.error('AI Drawer: Polling fallback error:', error);
|
||||
clearInterval(pollInterval);
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
@@ -468,38 +640,6 @@ class AIDrawer {
|
||||
}, 300000);
|
||||
}
|
||||
|
||||
// Метод для проверки результата
|
||||
async checkAIResult(taskId) {
|
||||
console.log('AI Drawer: Checking result for task:', taskId);
|
||||
|
||||
const response = await fetch(`/get_ai_result.php?task_id=${taskId}`);
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Result check failed: ${response.status}`);
|
||||
}
|
||||
|
||||
const data = await response.json();
|
||||
console.log('AI Drawer: Result check response:', data);
|
||||
|
||||
if (data.status === 'completed') {
|
||||
// Задача завершена
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage(data.response, false, 25);
|
||||
return true; // Остановить polling
|
||||
} else if (data.status === 'error') {
|
||||
// Ошибка обработки
|
||||
this.hideLoading();
|
||||
this.hideTypingIndicator();
|
||||
this.addStreamingMessage(data.error || 'Произошла ошибка при обработке запроса.', false, 25);
|
||||
return true; // Остановить polling
|
||||
} else {
|
||||
// Задача еще обрабатывается
|
||||
console.log('AI Drawer: Still processing...');
|
||||
return false; // Продолжить polling
|
||||
}
|
||||
}
|
||||
|
||||
getCurrentContext() {
|
||||
const urlParams = new URLSearchParams(window.location.search);
|
||||
const projectId = urlParams.get('record') || '';
|
||||
|
||||
Reference in New Issue
Block a user