Отправленные сервером события для уведомлений о решении CAPTCHA в реальном времени

Если страница в браузере ждёт решённый токен, самый дешёвый способ его доставить — не опрашивать res.php из клиента, а держать одно открытое соединение Server-Sent Events (SSE) с собственным сервером. Сервер получает результат от CaptchaAI по pingback и тем же кадром отдаёт его в поток. Опрос раз в 5 секунд добавляет к каждой задаче половину интервала «мёртвого» ожидания и десятки лишних HTTP-запросов; SSE убирает и то, и другое.

Ниже — рабочая схема: как связать pingback с потоком SSE, код на Flask и Express, что ломается при горизонтальном масштабировании и как это чинится через Redis Pub/Sub.

Когда SSE уместнее опроса, а когда нет

SSE — односторонний канал: данные идут только от сервера к клиенту. Для результатов CAPTCHA этого достаточно — клиенту нечего дописывать в соединение после отправки задачи.

Обратная сторона: SSE отдаёт результат туда, где уже открыт браузер. Для CLI-утилиты или воркера проще обработать pingback напрямую либо положить результат в очередь — промежуточный поток там не нужен.

Схема: pingback CaptchaAI → ваш сервер → браузер

[Client] ← SSE stream ← [Your Server] ← Callback ← [CaptchaAI]
   ↓                          ↑
   Submit task → [CaptchaAI] ──┘ (pingback URL points to your server)

Последовательность шагов такая:

1. Клиент открывает постоянное HTTP-соединение с вашей конечной точкой SSE.
2. Клиент отправляет задачу в CaptchaAI, указав в параметре pingback адрес обратного вызова на вашем сервере.
3. CaptchaAI решает задачу и вызывает этот адрес с параметрами id и code.
4. Обработчик кладёт токен в очередь нужного клиента, а генератор SSE отдаёт его строкой event: captcha-solved.

Важная деталь: pingback не отменяет res.php, он лишь убирает опрос из горячего пути. Логику «если через N секунд токена нет — запросить res.php один раз» стоит оставить как страховку.

Шаг 1: сервер на Python (Flask)

Три маршрута: /events/<client_id> держит поток, /submit отправляет задачу с pingback, /callback принимает результат и будит нужную очередь.

import os
import queue
import threading
import requests
from flask import Flask, Response, request, jsonify

app = Flask(__name__)

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# Per-client event queues: client_id -> Queue
client_queues = {}
queues_lock = threading.Lock()


@app.route("/events/<client_id>")
def sse_stream(client_id):
    """SSE endpoint — clients connect here for real-time results."""
    q = queue.Queue()

    with queues_lock:
        client_queues[client_id] = q

    def generate():
        try:
            while True:
                # Block until a result arrives (timeout for keepalive)
                try:
                    data = q.get(timeout=30)
                    yield f"event: captcha-solved\ndata: {data}\n\n"
                except queue.Empty:
                    # Send keepalive comment to prevent connection timeout
                    yield ": keepalive\n\n"
        finally:
            with queues_lock:
                client_queues.pop(client_id, None)

    return Response(
        generate(),
        mimetype="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # Disable nginx buffering
        }
    )


@app.route("/submit", methods=["POST"])
def submit_captcha():
    """Submit a CAPTCHA task with callback to this server."""
    data = request.json
    client_id = data["client_id"]
    sitekey = data["sitekey"]
    pageurl = data["pageurl"]

    callback_url = f"{request.host_url}callback?client_id={client_id}"

    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "pingback": callback_url,
        "json": 1
    })
    result = resp.json()

    if result.get("status") == 1:
        return jsonify({"task_id": result["request"]})
    return jsonify({"error": result.get("request")}), 400


@app.route("/callback")
def captcha_callback():
    """Receive CaptchaAI callback and push to SSE stream."""
    client_id = request.args.get("client_id")
    task_id = request.args.get("id")
    solution = request.args.get("code")

    import json
    message = json.dumps({
        "task_id": task_id,
        "solution": solution
    })

    with queues_lock:
        q = client_queues.get(client_id)
        if q:
            q.put(message)

    return "OK", 200


if __name__ == "__main__":
    app.run(port=5000, threaded=True)

Три места, на которых обычно спотыкаются:

• mimetype="text/event-stream" обязателен, иначе EventSource не распознает ответ.
• X-Accel-Buffering: no отключает буферизацию nginx. Без него события копятся и приходят пачкой.
• Каждое событие завершается двумя переводами строки — одиночный браузер не считает концом кадра.

Тайм-аут 30 секунд в q.get() нужен не для логики, а для keepalive: комментарий : keepalive держит соединение живым через прокси, которые рвут «молчащие» коннекты.

Шаг 2: клиент в браузере

<!DOCTYPE html>
<html>
<body>
  <button onclick="submitCaptcha()">Solve CAPTCHA</button>
  <div id="results"></div>

  <script>
    const clientId = crypto.randomUUID();
    const resultsDiv = document.getElementById("results");

    // Connect SSE stream
    const eventSource = new EventSource(`/events/${clientId}`);

    eventSource.addEventListener("captcha-solved", (event) => {
      const data = JSON.parse(event.data);
      resultsDiv.innerHTML += `<p>Task ${data.task_id}: ${data.solution.substring(0, 30)}...</p>`;
    });

    eventSource.onerror = () => {
      console.log("SSE connection lost, reconnecting...");
    };

    async function submitCaptcha() {
      const response = await fetch("/submit", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          client_id: clientId,
          sitekey: "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
          pageurl: "https://example.com"
        })
      });
      const result = await response.json();
      resultsDiv.innerHTML += `<p>Submitted: ${result.task_id}</p>`;
    }
  </script>
</body>
</html>

EventSource переподключается сам, поэтому onerror здесь только логирует. Собственную логику восстановления состояния вешайте на переподключение, а не на первую ошибку.

Шаг 3: тот же сервер на Node.js (Express)

const express = require("express");
const axios = require("axios");

const app = express();
app.use(express.json());

const API_KEY = process.env.CAPTCHAAI_API_KEY;
const BASE_URL = process.env.BASE_URL || "http://localhost:3000";

// Per-client SSE connections: clientId -> Response object
const clients = new Map();

// SSE endpoint
app.get("/events/:clientId", (req, res) => {
  const clientId = req.params.clientId;

  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    Connection: "keep-alive",
    "X-Accel-Buffering": "no",
  });

  clients.set(clientId, res);

  // Keepalive every 30 seconds
  const keepalive = setInterval(() => {
    res.write(": keepalive\n\n");
  }, 30000);

  req.on("close", () => {
    clearInterval(keepalive);
    clients.delete(clientId);
  });
});

// Submit CAPTCHA
app.post("/submit", async (req, res) => {
  const { client_id, sitekey, pageurl } = req.body;
  const callbackUrl = `${BASE_URL}/callback?client_id=${client_id}`;

  try {
    const resp = await axios.post("https://ocr.captchaai.com/in.php", null, {
      params: {
        key: API_KEY,
        method: "userrecaptcha",
        googlekey: sitekey,
        pageurl: pageurl,
        pingback: callbackUrl,
        json: 1,
      },
    });

    if (resp.data.status === 1) {
      return res.json({ task_id: resp.data.request });
    }
    res.status(400).json({ error: resp.data.request });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

// CaptchaAI callback → push to SSE
app.get("/callback", (req, res) => {
  const clientId = req.query.client_id;
  const taskId = req.query.id;
  const solution = req.query.code;

  const clientRes = clients.get(clientId);
  if (clientRes) {
    const data = JSON.stringify({ task_id: taskId, solution: solution });
    clientRes.write(`event: captcha-solved\ndata: ${data}\n\n`);
  }

  res.sendStatus(200);
});

app.listen(3000, () => console.log("SSE server running on :3000"));

Отличие от Flask-версии: объект res хранится прямо в Map, и в него пишет обработчик обратного вызова. Обязательно снимайте setInterval в req.on("close") — иначе на каждое оборванное соединение остаётся висящий таймер, и за сутки процесс съедает память на ровном месте.

Шаг 4: несколько экземпляров сервера

Главная проблема в эксплуатации: соединение SSE хранит состояние в конкретном процессе. Если за балансировщиком стоят три экземпляра, обратный вызов прилетит на случайный из них — и в двух случаях из трёх очереди клиента там просто нет.

Решение — шина сообщений. Обработчик обратного вызова публикует результат в Redis, а генератор SSE подписан на канал своего клиента:

# Callback handler publishes to Redis
import redis
r = redis.Redis()
r.publish(f"captcha:{client_id}", json.dumps(message))

# SSE handler subscribes to Redis
pubsub = r.pubsub()
pubsub.subscribe(f"captcha:{client_id}")
for msg in pubsub.listen():
    if msg["type"] == "message":
        yield f"data: {msg['data'].decode()}\n\n"

Тот же приём работает с любым уже развёрнутым брокером: Kafka, NATS, Postgres LISTEN/NOTIFY. Redis выбирают потому, что он почти всегда есть рядом.

Второе ограничение — браузерное: по HTTP/1.1 браузер держит не больше шести соединений на домен, и поток SSE занимает одно из них. Поэтому на клиента открывают ровно один поток и мультиплексируют в него результаты всех задач, различая их по task_id. По HTTP/2 лимит фактически снимается.

Практический пример: панель QA-прогонов в распределённой команде

Типичный сценарий для русскоязычных команд: тестировщики в Москве, Алматы и Минске работают с одной внутренней панелью прогонов, развёрнутой в европейском регионе облака. RTT у части команды — 60–90 мс, у части заметно больше, а мобильный интернет добавляет разрывы.

Опрос в такой топологии особенно неприятен: каждый тестировщик генерирует свой поток запросов к res.php, и при десяти открытых вкладках это сотни бесполезных обращений в минуту. С SSE каждая вкладка держит одно соединение, а токен reCAPTCHA v2 приходит в панель сразу после решения.

Тарификация считается по потокам, а не по вкладкам: план STANDARD ($30/мес, 15 потоков) закрывает пятнадцать одновременных задач, ADVANCE ($90/мес, 50 потоков) — пятьдесят. Число открытых соединений SSE на цену не влияет: это ваши HTTP-коннекты, а не задачи CaptchaAI.

Отдельная оговорка про логи. Если панель пишет историю прогонов, помните про 152-ФЗ «О персональных данных» и сохраняйте только те поля, которые вы вправе обрабатывать: task_id, тип задачи, время решения. Сам токен и учётные данные тестовых аккаунтов в долгоживущих логах не нужны.

Диагностика типичных сбоев

Отдельно про Cloudflare: поток SSE через него проходит, но ответы могут буферизоваться — помогает тот же X-Accel-Buffering: no либо явное отключение буферизации на маршруте.

Частые вопросы

Нужно ли всё равно опрашивать res.php, если настроен pingback?

В горячем пути — нет. Но обратный вызов может не дойти из-за сетевого сбоя или деплоя, поэтому оставьте резервную проверку: нет токена через разумный тайм-аут — запросите res.php один раз по сохранённому ID задачи.

Сколько одновременных соединений SSE выдержит один сервер?

Node.js спокойно держит десятки тысяч соединений: каждое — лёгкий keep-alive коннект. Flask с потоками упирается раньше — под высокую конкуренцию берите асинхронный стек (FastAPI с asyncio) или несколько воркеров за балансировщиком плюс Redis.

Что произойдёт при обрыве связи у клиента?

EventSource переподключится сам, но события, отправленные во время разрыва, теряются — SSE не хранит историю. Если это критично, кладите результат в Redis с TTL и отдавайте накопленное после переподключения.

Работает ли эта схема для GeeTest v3 и Cloudflare Turnstile?

Да. Параметр pingback не зависит от типа задачи: меняются только method и набор параметров при отправке. Схема доставки результата одна для всех поддерживаемых типов.

Подходит ли SSE для воркеров и скриптов без браузера?

Как правило, нет: для backend-процессов проще принять pingback напрямую или положить результат в очередь. SSE окупается там, где результат нужно показать в открытой странице.

Что почитать дальше

• быстрый старт CaptchaAI: первый решённый токен за пять минут
• решение reCAPTCHA v2 через API по шагам
• интеграция Cloudflare Turnstile через API
• работа с GeeTest v3 через API