DynamoDB para rastreamento de solução CAPTCHA sem servidor (guia completo)

O DynamoDB se adapta naturalmente aos fluxos de trabalho CAPTCHA sem servidor - sem dores de cabeça no pool de conexões, TTL integrado para limpeza automática e desempenho consistente em qualquer escala. Este guia cobre design de tabelas, estrutura de itens e padrões de consulta para rastrear soluções CAPTCHA em arquiteturas baseadas em Lambda.

Design de mesa

Padrão de tabela única

Uma tabela do DynamoDB lida com histórico de soluções, tarefas ativas e estatísticas agregadas:

Chave de partição (PK)	Chave de classificação (SK)	Objetivo
`SOLVE#{captcha_id}`	`META`	Resolver registro
`SITE#{sitekey}`	`SOLVE#{timestamp}`	Histórico de soluções por site
`STATS#{date}`	`TYPE#{captcha_type}`	Estatísticas agregadas diárias
`ACTIVE#{captcha_id}`	`TASK`	Rastreamento de tarefas durante o voo

Definição de tabela

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

Implementação Python

Configuração

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

Resolva e rastreie

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

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 de JavaScript

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

Otimização de custos

Estratégia	Impacto
Use o faturamento sob demanda para cargas de trabalho variáveis	Sem provisionamento excessivo
Habilite TTL para limpeza automática de registros	Reduz custos de armazenamento
O projeto precisava apenas de atributos nas consultas	Menor consumo de unidade de leitura
Gravações em lote com `BatchWriteItem`	Menos chamadas de API
Use fluxos do DynamoDB para análises	Descarregar agregação para Lambda

Solução de problemas

Problema	Causa	Correção
`ProvisionedThroughputExceededException`	Muitas gravações por segundo	Mude para o faturamento sob demanda ou aumente o WCU
Itens TTL não excluídos imediatamente	A exclusão do TTL do DynamoDB é eventual (aproximadamente 48 horas)	Não confie no TTL para limpeza em tempo real; filtrar itens expirados em consultas
Partição quente em `STATS#{date}`	Todos os trabalhadores gravando na mesma partição	Use sufixo aleatório: `STATS#{date}#shard{0-9}`
A consulta retorna muitos itens	Chave de partição ampla	Adicione condições SK para restringir resultados

Perguntas frequentes

Por que DynamoDB em vez de RDS para rastreamento CAPTCHA sem servidor?

O DynamoDB não tem limite de conexão – perfeito para Lambda, onde cada invocação abre uma nova conexão. O RDS requer pool de conexões (RDS Proxy), o que adiciona custo e complexidade.

Quanto custa o DynamoDB para rastreamento CAPTCHA?

Com faturamento sob demanda: aproximadamente US$ 1,25 por milhão de gravações e aproximadamente US$ 0,25 por milhão de leituras. Com 10.000 soluções/day, espere menos de US$ 1/month para armazenamento e acesso.

Posso consultar todos os tipos de CAPTCHA?

Use o índice GSI1 para consultar por status entre tipos. Para análises de tipo cruzado, agregue usando DynamoDB Streams e uma função Lambda que grava na partição STATS#.

Próximas etapas

Crie rastreamento CAPTCHA sem servidor que escala automaticamenteobtenha sua chave API CaptchaAI.

Guias relacionados:

AWS Lambda + CaptchaAI
História do MongoDB CAPTCHA
Gerenciamento de token TTL Redis

DynamoDB para rastreamento de solução CAPTCHA sem servidor

Design de mesa

Padrão de tabela única

Definição de tabela

Implementação Python

Configuração

Resolva e rastreie

Padrões de consulta

Implementação de JavaScript

Otimização de custos

Solução de problemas

Perguntas frequentes

Por que DynamoDB em vez de RDS para rastreamento CAPTCHA sem servidor?

Quanto custa o DynamoDB para rastreamento CAPTCHA?

Posso consultar todos os tipos de CAPTCHA?

Próximas etapas

Construindo soluções CAPTCHA orientadas a eventos com AWS SNS e CaptchaAI

Playbooks Ansible para implantação de trabalhador CaptchaAI

Impacto da resolução DNS no desempenho da API CAPTCHA

API CaptchaAI JSON vs API de formulário: qual formato usar

Trabalhadores de resolução de CAPTCHA com escalonamento automático

AWS Lambda + CaptchaAI: solução de CAPTCHA sem servidor

Design de mesa

Padrão de tabela única

Definição de tabela

Implementação Python

Configuração

Resolva e rastreie

Padrões de consulta

Implementação de JavaScript

Otimização de custos

Solução de problemas

Perguntas frequentes

Por que DynamoDB em vez de RDS para rastreamento CAPTCHA sem servidor?

Quanto custa o DynamoDB para rastreamento CAPTCHA?

Posso consultar todos os tipos de CAPTCHA?

Próximas etapas

Postagens relacionadas

Construindo soluções CAPTCHA orientadas a eventos com AWS SNS e CaptchaAI

Playbooks Ansible para implantação de trabalhador CaptchaAI

Impacto da resolução DNS no desempenho da API CAPTCHA

API CaptchaAI JSON vs API de formulário: qual formato usar

Trabalhadores de resolução de CAPTCHA com escalonamento automático

AWS Lambda + CaptchaAI: solução de CAPTCHA sem servidor