DynamoDB для отслеживания решения бессерверной CAPTCHA

В Lambda нет постоянного соединения с базой между вызовами, поэтому пул соединений PostgreSQL или MySQL скорее мешает. DynamoDB снимает проблему: HTTPS-подключение на каждый вызов, встроенный TTL и стабильная скорость при всплесках нагрузки.

Почему не реляционная база данных

В Lambda каждый вызов может открыть новое подключение, а PostgreSQL и MySQL ограничены числом одновременных соединений на инстанс. При резком всплеске трафика — например, когда десятки Lambda параллельно отправляют задачи в in.php и опрашивают res.php — этот лимит быстро исчерпывается, и приходится добавлять RDS Proxy: отдельный платный компонент и ещё одна точка отказа. DynamoDB обращается по HTTPS на каждый запрос, поэтому лимита соединений нет вовсе, а PAY_PER_REQUEST снимает вопрос заранее рассчитанной пропускной способности.

Проектирование таблицы

Паттерн «одна таблица на всё»

Одна таблица закрывает историю решений, активные задачи и дневную статистику — single-table design без join'ов:

Такое разбиение ключей закрывает четыре сценария одним и тем же индексом:

• По ID решения — быстрый lookup статуса конкретной задачи (SOLVE#{captcha_id}).
• По sitekey — история решений для конкретного сайта, отсортированная по времени.
• По дате — дневная статистика без Scan по всей таблице.
• По активной задаче — TTL-запись, которая сама исчезает через 10 минут, если решение зависло.

Схема таблицы

{
  "TableName": "CaptchaSolves",
  "KeySchema": [
    { "AttributeName": "PK", "KeyType": "HASH" },
    { "AttributeName": "SK", "KeyType": "RANGE" }
  ],
  "AttributeDefinitions": [
    { "AttributeName": "PK", "KeyType": "S" },
    { "AttributeName": "SK", "KeyType": "S" },
    { "AttributeName": "GSI1PK", "KeyType": "S" },
    { "AttributeName": "GSI1SK", "KeyType": "S" }
  ],
  "GlobalSecondaryIndexes": [
    {
      "IndexName": "GSI1",
      "KeySchema": [
        { "AttributeName": "GSI1PK", "KeyType": "HASH" },
        { "AttributeName": "GSI1SK", "KeyType": "RANGE" }
      ],
      "Projection": { "ProjectionType": "ALL" }
    }
  ],
  "BillingMode": "PAY_PER_REQUEST",
  "TimeToLiveSpecification": {
    "AttributeName": "ttl",
    "Enabled": true
  }
}

BillingMode: PAY_PER_REQUEST (on-demand) не требует заранее угадывать нагрузку — удобно при нерегулярных триггерах Lambda.

Реализация на Python

Настройка окружения

import os
import time
from datetime import datetime, timezone
import boto3
import requests

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ.get("DYNAMODB_TABLE", "CaptchaSolves"))
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

Функция solve_and_track: отправка, опрос и запись результата

Отправляет задачу в in.php, опрашивает res.php каждые 5 секунд (до 60 попыток — укладывается в лимит выполнения Lambda) и на каждом исходе пишет отдельную запись: активную задачу, итог по sitekey и обновление дневной статистики. Поле GSI1PK дублирует статус решения, чтобы get_active_tasks мог выбрать задачи в процессе без Scan по всей таблице:

def solve_and_track(sitekey, pageurl, captcha_type="recaptcha_v2", project=None):
    now = datetime.now(timezone.utc)
    timestamp = now.isoformat()
    ttl_90_days = int(now.timestamp()) + (90 * 24 * 3600)

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

    if data.get("status") != 1:
        # Store error record
        table.put_item(Item={
            "PK": f"SITE#{sitekey}",
            "SK": f"SOLVE#{timestamp}",
            "captcha_type": captcha_type,
            "pageurl": pageurl,
            "status": "error",
            "error": data.get("request"),
            "submitted_at": timestamp,
            "project": project or "default",
            "ttl": ttl_90_days,
            "GSI1PK": f"STATUS#error",
            "GSI1SK": timestamp
        })
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Track active task
    table.put_item(Item={
        "PK": f"ACTIVE#{captcha_id}",
        "SK": "TASK",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "captcha_type": captcha_type,
        "submitted_at": timestamp,
        "ttl": int(now.timestamp()) + 600  # Auto-clean in 10 min
    })

    # Poll for result
    polls = 0
    for _ in range(60):
        time.sleep(5)
        polls += 1
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get",
            "id": captcha_id, "json": 1
        }).json()

        if result.get("status") == 1:
            solved_at = datetime.now(timezone.utc).isoformat()
            elapsed_ms = int(
                (datetime.now(timezone.utc) - now).total_seconds() * 1000
            )

            # Store success record
            table.put_item(Item={
                "PK": f"SOLVE#{captcha_id}",
                "SK": "META",
                "captcha_type": captcha_type,
                "sitekey": sitekey,
                "pageurl": pageurl,
                "status": "solved",
                "submitted_at": timestamp,
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls,
                "project": project or "default",
                "ttl": ttl_90_days,
                "GSI1PK": f"STATUS#solved",
                "GSI1SK": timestamp
            })

            # Also store in site history
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "solved",
                "elapsed_ms": elapsed_ms,
                "ttl": ttl_90_days
            })

            # Remove active task
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })

            # Update daily stats
            update_daily_stats(captcha_type, True, elapsed_ms)

            return {"solution": result["request"]}

        if result.get("request") != "CAPCHA_NOT_READY":
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "error",
                "error": result.get("request"),
                "ttl": ttl_90_days
            })
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })
            update_daily_stats(captcha_type, False, 0)
            return {"error": result.get("request")}

    table.delete_item(Key={"PK": f"ACTIVE#{captcha_id}", "SK": "TASK"})
    update_daily_stats(captcha_type, False, 0)
    return {"error": "TIMEOUT"}


def update_daily_stats(captcha_type, success, elapsed_ms):
    date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    update_expr = "SET total_solves = if_not_exists(total_solves, :zero) + :one"
    expr_values = {":zero": 0, ":one": 1}

    if success:
        update_expr += ", successful = if_not_exists(successful, :zero) + :one"
        update_expr += ", total_elapsed = if_not_exists(total_elapsed, :zero) + :elapsed"
        expr_values[":elapsed"] = elapsed_ms
    else:
        update_expr += ", failed = if_not_exists(failed, :zero) + :one"

    table.update_item(
        Key={"PK": f"STATS#{date_str}", "SK": f"TYPE#{captcha_type}"},
        UpdateExpression=update_expr,
        ExpressionAttributeValues=expr_values
    )

Готовые запросы

История по sitekey, статистика за день и активные задачи — все три запроса опираются на один и тот же GSI1, поэтому выборка по статусу не требует Scan и дополнительного индекса:

def get_site_history(sitekey, limit=50):
    """Get recent solves for a specific site key."""
    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"SITE#{sitekey}"},
        ScanIndexForward=False,
        Limit=limit
    )
    return response["Items"]


def get_daily_stats(date_str=None):
    """Get stats for a specific date (default: today)."""
    if not date_str:
        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"STATS#{date_str}"}
    )
    return response["Items"]


def get_active_tasks():
    """List all currently active CAPTCHA tasks."""
    response = table.query(
        IndexName="GSI1",
        KeyConditionExpression="GSI1PK = :pk",
        ExpressionAttributeValues={":pk": "STATUS#polling"}
    )
    return response["Items"]

Оптимизация расходов

При тарификации on-demand DynamoDB стоит ориентировочно $1,25 за миллион операций записи и $0,25 за миллион операций чтения. При 10 000 решённых CAPTCHA в день на каждое решение приходится по 3–5 операций записи (активная задача, запись решения, история по sitekey, обновление статистики) — это заметно дешевле $1 в месяц на хранение и доступ. TTL и точечные GSI-запросы держат объём таблицы предсказуемым, поэтому пересчитывать расходы вручную не нужно.

Реализация на Node.js

Та же логика на @aws-sdk/lib-dynamodb (SDK v3):

const { DynamoDBClient } = require("@aws-sdk/client-dynamodb");
const { DynamoDBDocumentClient, PutCommand, QueryCommand, UpdateCommand } = require("@aws-sdk/lib-dynamodb");
const axios = require("axios");

const client = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const TABLE = process.env.DYNAMODB_TABLE || "CaptchaSolves";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

async function solveAndTrack(sitekey, pageurl, type = "recaptcha_v2") {
  const now = new Date();
  const timestamp = now.toISOString();
  const ttl = Math.floor(now.getTime() / 1000) + 90 * 24 * 3600;

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

  if (submit.data.status !== 1) {
    await client.send(new PutCommand({
      TableName: TABLE,
      Item: { PK: `SITE#${sitekey}`, SK: `SOLVE#${timestamp}`, status: "error", error: submit.data.request, ttl },
    }));
    return { error: submit.data.request };
  }

  const captchaId = submit.data.request;
  let polls = 0;

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      const elapsed = Date.now() - now.getTime();
      await client.send(new PutCommand({
        TableName: TABLE,
        Item: {
          PK: `SOLVE#${captchaId}`, SK: "META", captcha_type: type,
          sitekey, pageurl, status: "solved", submitted_at: timestamp,
          solved_at: new Date().toISOString(), elapsed_ms: elapsed, polls, ttl,
        },
      }));
      return { solution: poll.data.request };
    }

    if (poll.data.request !== "CAPCHA_NOT_READY") {
      return { error: poll.data.request };
    }
  }
  return { error: "TIMEOUT" };
}

async function getSiteHistory(sitekey, limit = 50) {
  const result = await client.send(new QueryCommand({
    TableName: TABLE,
    KeyConditionExpression: "PK = :pk",
    ExpressionAttributeValues: { ":pk": `SITE#${sitekey}` },
    ScanIndexForward: false,
    Limit: limit,
  }));
  return result.Items;
}

Типичные проблемы и их решение

Ниже — ошибки, с которыми чаще всего сталкиваются при переходе с provisioned-режима или при первом всплеске нагрузки:

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

Вопросы, которые обычно возникают уже после первого разворачивания такой схемы в проде:

Можно ли использовать одну таблицу для нескольких проектов?

Да — в схеме уже есть атрибут project (см. project or "default" в коде). Добавьте его в PK, например PROJECT#{project}#SITE#{sitekey}, и одна таблица обслужит нескольких клиентов без лишних расходов.

Как получить статистику по всем типам CAPTCHA и ловить падение успешности?

GSI1 даёт выборку по статусу вне зависимости от типа. DynamoDB Streams + Lambda-агрегатор в STATS#{date} — сводка по reCAPTCHA/Turnstile/GeeTest v3 и алерт при просадке successful/total_solves без сканирования таблицы.

Что учитывать по 152-ФЗ, если в таблице хранятся pageurl и sitekey?

sitekey и pageurl сами по себе не персональные данные, но если в запись попадает логин или email из формы, для аудитории в РФ это уже 152-ФЗ, а для трансграничных команд — due diligence в духе GDPR. Храните только то, что нужно для отладки.

В каком регионе AWS размещать таблицу, если команда работает из СНГ?

DynamoDB — региональный сервис: держите таблицу в том же регионе, что и Lambda, иначе к времени решения CAPTCHA добавится межрегиональная сетевая задержка. Команды из России, Казахстана и Центральной Азии чаще всего берут eu-central-1 (Франкфурт) как ближайший регион AWS с полным набором сервисов. Global Tables для этой нагрузки избыточны — записи короткоживущие, и TTL удаляет их сам.

Следующие шаги

• CaptchaAI Quickstart: ваше первое решение CAPTCHA за 5 минут
• Как решить reCAPTCHA v2 через API: пошаговое руководство
• Как решить Cloudflare Turnstile через API
• Как решить GeeTest v3 с помощью API