DynamoDB pour le suivi de résolution de CAPTCHA sans serveur

Pour tracer vos résolutions de CAPTCHA dans une pile serverless, DynamoDB est le magasin le plus simple à opérer : aucune connexion à gérer, un TTL natif qui purge les vieux enregistrements et une latence stable quelle que soit la charge. C'est ce qui manque à une base relationnelle sous Lambda, où chaque invocation doit rouvrir sa connexion. Voici ce que vous mettez en place avec l'API CaptchaAI :

• une table unique qui porte l'historique, les tâches en cours et les statistiques ;
• un TTL qui purge automatiquement les enregistrements arrivés à expiration ;
• des requêtes prêtes à alimenter un tableau de bord de suivi.

Modéliser la table en single-table

Une seule table porte l'historique des résolutions, les tâches en cours et les statistiques agrégées. Le couple clé de partition / clé de tri encode chaque type d'accès, ce qui évite de multiplier les tables :

Définir le schéma et l'index secondaire

L'index secondaire global GSI1 filtre par statut sans balayer toute la table, et PAY_PER_REQUEST évite le surprovisionnement sur un trafic irrégulier :

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

Configurer le client en Python

Le client boto3 récupère ses identifiants via le rôle IAM de la fonction ; seuls le nom de la table et la clé API CaptchaAI viennent des variables d'environnement :

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

Résoudre puis enregistrer chaque tâche

La fonction envoie la tâche à CaptchaAI, écrit un enregistrement ACTIVE# à TTL court, puis interroge le résultat toutes les 5 secondes. Dès qu'une résolution aboutit, elle mesure le temps écoulé, journalise le succès et retire la tâche en vol. Les enregistrements finaux portent un TTL à 90 jours :

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
    )

Interroger l'historique et les statistiques

Trois accès couvrent l'essentiel du suivi, sans jamais balayer la table entière :

• get_site_history renvoie les dernières résolutions d'un site, les plus récentes en tête grâce à ScanIndexForward=False ;
• get_daily_stats agrège les compteurs d'une journée ;
• get_active_tasks liste les tâches encore en vol via 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"]

Reproduire le suivi en Node.js

Sous Node.js, le SDK v3 (@aws-sdk/lib-dynamodb) reproduit la même logique et la même structure d'éléments qu'en Python, ce qui garde vos requêtes interchangeables entre les deux runtimes :

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

Maîtriser les coûts DynamoDB

La facturation à la demande reste le choix par défaut sur un trafic irrégulier : vous ne payez que les lectures et écritures réelles, et le TTL contient la facture à fort volume. Si votre worker vise une région comme eu-west-3 (Paris), gardez la table dans la même région que la fonction Lambda pour éviter la latence inter-région.

Dépannage

Les incidents les plus fréquents viennent du débit d'écriture et du comportement différé du TTL :

Questions fréquentes

Les points qui reviennent le plus souvent avant de passer en production :

Le TTL DynamoDB supprime-t-il vraiment les enregistrements à l'heure prévue ?

Pas exactement. DynamoDB supprime les éléments expirés de façon différée, dans un délai qui peut atteindre 48 heures. Le TTL contient les coûts de stockage, sans purge à la seconde près. Filtrez donc les enregistrements expirés dans vos requêtes, avec une condition sur l'attribut ttl.

DynamoDB convient-il mieux que RDS pour un worker Lambda ?

Dans la plupart des cas, oui. DynamoDB n'impose aucune limite de connexion, ce qui colle au modèle Lambda où chaque invocation ouvre la sienne. RDS exige RDS Proxy pour mutualiser les connexions et ne reprend l'avantage que pour des jointures relationnelles complexes.

Comment éviter une partition chaude sur les statistiques quotidiennes ?

Le point sensible est la clé STATS#{date} : le même jour, tous les workers écrivent sur la même partition et le débit sature. Ajoutez un suffixe aléatoire, par exemple STATS#{date}#shard{0-9}, puis additionnez les dix shards à la lecture. Les écritures se répartissent alors sur plusieurs partitions physiques.

Que stocker sans risque côté RGPD ?

Minimisez les données personnelles. Une pageurl ou un sitekey restent techniques, mais évitez d'y joindre des identifiants d'utilisateur en clair. Gardez un TTL court pour ne pas conserver l'historique indéfiniment et vérifiez vos obligations RGPD sur la rétention.

Pour aller plus loin

Montez un suivi serverless qui s'adapte à la charge : récupérez votre clé API CaptchaAI et enregistrez votre première résolution.

Guides associés :

• déployer CaptchaAI sur AWS Lambda
• l'historique des résolutions avec MongoDB
• gérer le TTL des tokens Redis