Tutorials

DynamoDB para rastreamento de solução CAPTCHA sem servidor

Automação Python Todos os tipos de CAPTCHA
Por Equipe CaptchaAI Brasil
Publicado em Jun 02, 2026
DynamoDB para rastreamento de solução CAPTCHA sem servidor
Imagem em destaque indisponível

Se os workers de CAPTCHA rodam em AWS Lambda, cada invocação abre sua própria conexão - e é aí que um banco relacional emperra. A resposta é o DynamoDB: sem pool de conexões, TTL nativo e latência estável em qualquer volume, seja em us-east-1 ou em sa-east-1 (São Paulo). Este guia cobre design de tabela, estrutura de itens e padrões de consulta para rastrear CAPTCHA em arquiteturas Lambda, com exemplos em Python e Node.js.

Como estruturar a tabela no DynamoDB

Um único padrão de tabela para tudo

Uma única tabela guarda o histórico de soluções, as tarefas em andamento e as estatísticas diárias. Esse "single-table design" evita joins e mantém o custo previsível:

Chave de partição (PK) Chave de classificação (SK) Finalidade
SOLVE#{captcha_id} META Registro da solução
SITE#{sitekey} SOLVE#{timestamp} Histórico de soluções por site
STATS#{date} TYPE#{captcha_type} Estatísticas diárias agregadas
ACTIVE#{captcha_id} TASK Tarefas em andamento

Definição da tabela em JSON

A definição cria a tabela com faturamento sob demanda e ativa o TTL no atributo ttl:

{
  "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
  }
}

Quando esse padrão não é a melhor escolha

Nem todo workflow de CAPTCHA se beneficia do single-table design. Vale considerar uma alternativa quando:

  • As consultas exigem joins complexos entre entidades — por exemplo, cruzar o histórico de solução com dados de faturamento no mesmo relatório. Um banco relacional com RDS Proxy resolve isso com menos código na aplicação.
  • A equipe já opera um cluster PostgreSQL ou MySQL para o resto do sistema e prefere não manter dois paradigmas de dados em produção.
  • É preciso consistência forte entre múltiplas regiões AWS ao mesmo tempo. O DynamoDB Global Tables cobre esse caso, mas soma uma camada de operação que nem todo time precisa.

Para a maioria dos workers Lambda que apenas gravam e consultam resultados de CAPTCHA, no entanto, o DynamoDB continua sendo a opção mais simples de operar.

Implementação em Python

Configuração inicial

Aponte o client do boto3 para a tabela e carregue a chave de API da CaptchaAI a partir do ambiente:

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"]

Função para resolver e gravar o rastreamento

A função envia o desafio à CaptchaAI, registra a tarefa como ativa, faz polling a cada 5 segundos e grava o registro final - sucesso ou erro - com TTL de 90 dias:

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
    )

Padrões de consulta mais usados

Três funções cobrem os casos mais comuns: histórico por site, estatísticas do dia e tarefas em polling:

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"]

Implementação em Node.js

O mesmo fluxo em Node.js, com o SDK v3 (@aws-sdk/lib-dynamodb) e axios:

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;
}

Como reduzir custos no DynamoDB

Cinco ajustes simples mantêm a fatura previsível mesmo em alto volume:

  • Faturamento sob demanda para cargas variáveis — evita provisionamento excessivo nos picos de tráfego.
  • TTL ativado em todo item transitório — reduz o custo de armazenamento sem rotina de limpeza manual.
  • Projeção de atributos nas consultas — retorne só os campos necessários para consumir menos unidades de leitura.
  • Gravações agrupadas com BatchWriteItem — menos chamadas de API por lote de resultados.
  • DynamoDB Streams para alimentar métricas — move a agregação para uma Lambda separada, fora do caminho crítico do worker.

Solução de problemas comuns

ProvisionedThroughputExceededException ao gravar

Acontece quando as gravações por segundo passam do limite provisionado. Migre para faturamento sob demanda ou aumente o WCU da tabela.

Itens com TTL continuam visíveis depois do prazo

A exclusão por TTL no DynamoDB é assíncrona e pode levar até ~48 horas. Não confie no TTL para limpeza em tempo real — filtre itens expirados diretamente nas consultas.

Partição "quente" em STATS#{date}

Todos os workers gravam na mesma partição no mesmo dia. Adicione um sufixo aleatório à chave, por exemplo STATS#{date}#shard{0-9}.

A consulta retorna itens demais

Sinal de chave de partição ampla demais. Adicione condições na SK para restringir o resultado.

Quando os registros trazem algum identificador do usuário, como IP ou sessão, o TTL de 90 dias também ajuda a manter a coleta alinhada ao princípio de minimização de dados da LGPD — avalie com sua equipe jurídica o prazo de retenção adequado ao seu caso de uso.

Perguntas frequentes

O TTL do DynamoDB apaga os registros de CAPTCHA na hora exata?

Não. A exclusão por TTL é assíncrona e pode levar até 48 horas. Se a limpeza em tempo real for crítica, filtre os itens vencidos direto na consulta.

Por que escolher o DynamoDB em vez do RDS para rastrear CAPTCHA a partir do Lambda?

O DynamoDB não impõe limite de conexões - cada invocação do Lambda grava direto, sem pool. O RDS exigiria o RDS Proxy, o que soma custo e complexidade.

Quanto custa manter o histórico de CAPTCHA no DynamoDB?

Com faturamento sob demanda: cerca de US$ 1,25 por milhão de gravações e US$ 0,25 por milhão de leituras. Em 10 mil soluções por dia, o custo de armazenamento e acesso fica abaixo de US$ 1 por mês.

Vale ativar o point-in-time recovery (PITR) nessa tabela?

Sim, para os itens SOLVE# e STATS#, que guardam o histórico de longo prazo — o PITR restaura a tabela para qualquer segundo dos últimos 35 dias, sem custo de operação contínua. Os itens ACTIVE#{captcha_id}, com TTL de 10 minutos, não precisam dessa proteção.

Faz sentido ativar o DynamoDB Streams só para gerar métricas?

Sim, principalmente com volume alto. O Streams tira a agregação do caminho crítico: uma Lambda separada processa os eventos e atualiza os contadores sem atrasar o retorno do token.

Próximos passos

Monte um rastreamento de CAPTCHA serverless que escala sozinho. Obtenha sua chave de API da CaptchaAI e grave a primeira solução hoje.

Guias relacionados:

Os comentários estão desativados para este artigo.