DynamoDB para seguimiento serverless de resoluciones CAPTCHA

En una arquitectura serverless, cada función Lambda que resuelve un CAPTCHA debería dejar un registro auditable: qué sitekey se procesó, cuánto tardó y si terminó en éxito o error. La base de datos que mejor encaja para ese registro es DynamoDB, y el motivo es directo: no hay conexiones que abrir ni cerrar en cada invocación. Sin connection pooling, con TTL nativo para la limpieza y con un rendimiento estable a cualquier volumen, se convierte en el almacén ideal para el historial de resoluciones sobre CaptchaAI. Esto es lo que vas a montar:

• Una tabla única con claves de partición y clasificación para todos los accesos.
• La función que resuelve el CAPTCHA con la API y escribe cada resultado.
• Las consultas de historial por sitio, estadísticas diarias y tareas activas.
• La limpieza automática por TTL y el control de costos.

Diseño de la tabla en DynamoDB

Modelo de tabla única (single-table design)

Una sola tabla de DynamoDB cubre tres necesidades a la vez: el historial de resoluciones, las tareas en curso y las estadísticas agregadas. Las combinaciones de clave de partición y clave de clasificación definen cada acceso:

En la práctica, cada prefijo cumple un papel concreto:

• SOLVE# guarda el registro maestro de cada resolución con su elapsed_ms.
• SITE# agrupa el historial por sitekey, ordenado por fecha.
• STATS# acumula los totales de cada día por tipo de CAPTCHA.
• ACTIVE# rastrea las tareas que siguen en vuelo, con un TTL corto.

Definición de la tabla y el índice GSI

La tabla usa facturación bajo demanda, un índice secundario global (GSI1) para consultar por estado y el atributo ttl activado para que DynamoDB caduque los registros antiguos por sí solo:

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

Implementación en Python

Preparar el cliente y la clave API

Primero, el recurso de DynamoDB y la clave API leídos desde variables de entorno. Nunca escribas tu API key directamente en el código:

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

Resolver el CAPTCHA y registrar el resultado

La función envía la tarea a CaptchaAI mediante in.php, guarda la tarea activa con un TTL corto y luego consulta el resultado en res.php. Cada desenlace (éxito, error o tiempo de espera agotado) queda escrito en la tabla y refleja en las estadísticas del día:

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
    )

Consultas: historial, estadísticas y tareas activas

Con la tabla poblada, cada patrón de acceso se traduce en una sola query, sin escaneos completos:

• get_site_history: últimas resoluciones de un sitekey.
• get_daily_stats: totales agregados de una fecha.
• get_active_tasks: tareas en curso vía GSI1.

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

Implementación en JavaScript (Node.js)

Si tu backend es Node.js, el SDK v3 de AWS ofrece la misma lógica con el DynamoDBDocumentClient. El flujo se reparte en tres pasos:

• Enviar la tarea a CaptchaAI con axios y recibir el captcha_id.
• Sondear res.php cada cinco segundos hasta obtener la solución.
• Escribir el registro de éxito con el elapsed_ms y el número de sondeos.

El resultado es equivalente al de Python, con la sintaxis asíncrona propia de Node.js:

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

Problemas frecuentes y cómo resolverlos

Estos son los tropiezos que aparecen al escalar el seguimiento a miles de resoluciones diarias, con su causa y su solución:

Cómo reducir el costo de DynamoDB

DynamoDB es económico para este caso de uso si aplicas un puñado de ajustes. El TTL elimina el gasto de almacenamiento a largo plazo y la facturación bajo demanda evita pagar capacidad que no usas:

Un caso práctico: reporte por cliente en una agencia

Piensa en una agencia de automatización en México o Argentina que ejecuta recolección de datos autorizada para varios clientes y factura en USD. La tabla que acabas de montar le sirve como fuente de verdad para el reporte mensual:

• Cada cliente corresponde a un valor de project distinto en cada registro.
• El historial por SITE#{sitekey} alimenta el resumen de resoluciones por portal.
• Las estadísticas de STATS#{date} dan el total diario y el tiempo medio.

Como CaptchaAI factura por thread —con resoluciones ilimitadas por thread dentro del mes— el costo del solver es predecible en dólares, y DynamoDB añade una capa de trazabilidad casi gratuita para justificar cada factura. Recuerda respetar siempre los términos de servicio de los sitios y la normativa de protección de datos aplicable.

Preguntas frecuentes

¿El TTL de DynamoDB borra los registros al instante?

No. La eliminación por TTL es eventual y puede tardar hasta unas 48 horas. Sirve muy bien para contener el almacenamiento a largo plazo, pero si necesitas que un registro deje de aparecer de inmediato, filtra los elementos caducados por el campo ttl dentro de la propia consulta.

¿Cómo evito una partición caliente en las estadísticas diarias?

El riesgo aparece cuando muchos workers escriben a la vez en STATS#{date}. Reparte la carga añadiendo un sufijo de shard aleatorio (STATS#{date}#shard{0-9}) y suma los diez shards al leer. Así distribuyes las escrituras en lugar de concentrarlas en una sola clave de partición.

¿Cuánto cuesta almacenar las resoluciones en DynamoDB?

Con facturación bajo demanda ronda los $1.25 por millón de escrituras y $0.25 por millón de lecturas. Con 10.000 resoluciones al día, el almacenamiento y el acceso suelen quedar por debajo de $1 al mes. El TTL a 90 días mantiene la tabla acotada.

¿Necesito una tabla distinta por cada tipo de CAPTCHA?

No. El modelo de tabla única cubre reCAPTCHA v2/v3, Turnstile, GeeTest v3 y el resto de tipos compatibles en la misma tabla; el atributo captcha_type los separa. Para análisis entre tipos, consulta por el índice GSI1 y agrega con DynamoDB Streams hacia la partición STATS#.

Guías relacionadas

• Resolver CAPTCHA sin servidor con AWS Lambda y CaptchaAI
• Guardar el historial de resoluciones en MongoDB
• Gestionar el TTL de tokens en Redis