godcrm/backend/routes/v3/telegramNikitron/polling.js

// backend/routes/v3/telegramNikitron/polling.js
// Agent triggering via HTTP and response polling/forwarding

import { apiLogger } from '../../../utils/logger.js';
import { dbGet, dbAll, isPostgres, safeJsonParse } from '../../../database/connection.js';
import { activeSessions } from './config.js';
import { sendMessage } from './shared.js';

export async function triggerAgentViaHttp(conversationId, content, adminUserId, attachments = []) {
  try {
    const baseUrl = process.env.INTERNAL_URL || 'http://localhost:' + (process.env.PORT || 5001);
    const { default: jwt } = await import('jsonwebtoken');
    const jwtSecret = process.env.JWT_SECRET;
    if (!jwtSecret) {
      apiLogger.error('[NikitronBot] JWT_SECRET not set');
      return false;
    }
    const token = jwt.sign({ id: adminUserId, email: 'nikitron@godcrm.local' }, jwtSecret, { expiresIn: '1h' });

    const response = await fetch(`${baseUrl}/api/v3/chat/conversations/${conversationId}/messages`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${token}`,
      },
      body: JSON.stringify({
        content,
        content_type: 'text',
        agent_mode: 'agent',
        attachments,
      }),
    });

    if (!response.ok) {
      const body = await response.text();
      apiLogger.error({ status: response.status, body, conversationId }, '[NikitronBot] Agent trigger failed');
      return false;
    }
    return true;
  } catch (err) {
    apiLogger.error({ err, conversationId }, '[NikitronBot] Agent trigger error');
    return false;
  }
}

export async function pollAndForwardAgentResponse(chatId, conversationId, afterMessageId, maxRetries = 90) {
  const FAST_POLL_MS = 2000;
  const SLOW_POLL_MS = 5000;
  let processingWentFalse = false;

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    const interval = attempt < 15 ? FAST_POLL_MS : SLOW_POLL_MS;
    await new Promise(resolve => setTimeout(resolve, interval));

    try {
      const conv = await dbGet(
        isPostgres()
          ? `SELECT is_processing FROM conversations WHERE id = $1`
          : `SELECT is_processing FROM conversations WHERE id = ?`,
        [conversationId]
      );

      const newMessages = await dbAll(
        isPostgres()
          ? `SELECT id, content, sender_type, role, metadata, created_at
             FROM messages
             WHERE conversation_id = $1 AND id > $2 AND role = 'assistant'
               AND content_type = 'text'
             ORDER BY id ASC`
          : `SELECT id, content, sender_type, role, metadata, created_at
             FROM messages
             WHERE conversation_id = ? AND id > ? AND role = 'assistant'
               AND content_type = 'text'
             ORDER BY id ASC`,
        [conversationId, afterMessageId]
      );

      if (newMessages.length > 0) {
        for (const msg of newMessages) {
          const metadata = safeJsonParse(msg.metadata, {});
          const agentName = metadata.agent_name || 'Agent';
          const text = (msg.content || '').trim();
          if (!text) continue;

          let displayText = text;
          if (displayText.length > 3900) {
            displayText = displayText.substring(0, 3900) + '\n\n... _(truncated)_';
          }

          await sendMessage(chatId, `*${agentName}:*\n\n${displayText}`);
        }

        const maxId = newMessages[newMessages.length - 1].id;
        const session = activeSessions.get(chatId);
        if (session) session.lastPolledMessageId = maxId;
        return true;
      }

      if (!conv?.is_processing) {
        if (processingWentFalse) return false;
        processingWentFalse = true;
      }
    } catch (err) {
      apiLogger.error({ err, conversationId, attempt }, '[NikitronBot] Poll error');
    }
  }

  // Timeout
  await sendMessage(chatId, 'Agent is taking longer than expected. Check CRM:\nhttps://devcrm.hltrn.cc/chat/' + conversationId);

  // Background follow-up (fire and forget)
  scheduleDelayedForward(chatId, conversationId, afterMessageId).catch(() => {});
  return false;
}

export async function scheduleDelayedForward(chatId, conversationId, afterMessageId) {
  for (let i = 0; i < 60; i++) {
    await new Promise(r => setTimeout(r, 10000));
    try {
      const newMessages = await dbAll(
        isPostgres()
          ? `SELECT id, content, metadata FROM messages
             WHERE conversation_id = $1 AND id > $2 AND role = 'assistant' AND content_type = 'text'
             ORDER BY id ASC`
          : `SELECT id, content, metadata FROM messages
             WHERE conversation_id = ? AND id > ? AND role = 'assistant' AND content_type = 'text'
             ORDER BY id ASC`,
        [conversationId, afterMessageId]
      );

      if (newMessages.length > 0) {
        for (const msg of newMessages) {
          const metadata = safeJsonParse(msg.metadata, {});
          const text = (msg.content || '').trim();
          if (!text) continue;
          let displayText = text.length > 3900 ? text.substring(0, 3900) + '\n\n...' : text;
          await sendMessage(chatId, `*${metadata.agent_name || 'Agent'}* _(delayed):_\n\n${displayText}`);
        }
        const session = activeSessions.get(chatId);
        if (session) session.lastPolledMessageId = newMessages[newMessages.length - 1].id;
        return;
      }

      const conv = await dbGet(
        isPostgres()
          ? `SELECT is_processing FROM conversations WHERE id = $1`
          : `SELECT is_processing FROM conversations WHERE id = ?`,
        [conversationId]
      );
      if (!conv?.is_processing && i > 5) return;
    } catch (_) {}
  }
}