DynamoDB untuk Pelacakan Penyelesaian CAPTCHA Tanpa Server

Berapa biaya sebenarnya untuk melacak ribuan penyelesaian CAPTCHA per hari di Lambda? Dengan DynamoDB dan billing on-demand, sebagian besar tim membayar di bawah $1 per bulan — karena tidak ada connection pool yang harus dikelola, TTL bawaan membersihkan data lama secara otomatis, dan performa kueri tetap stabil walau volume naik drastis. Panduan ini membahas desain tabel tunggal, kode Python dan Node.js yang siap pakai, sampai pola kueri untuk melacak setiap task CaptchaAI — mulai dari submit ke in.php, polling res.php, hingga token tersimpan rapi di DynamoDB.

Kenapa DynamoDB Cocok untuk Pelacakan CAPTCHA Serverless

Tim otomasi yang menjalankan Lambda di region seperti ap-southeast-1 (Singapura) atau ap-southeast-3 (Jakarta) biasanya menghadapi pola beban kerja yang sama: volume task CAPTCHA naik-turun tajam mengikuti jam operasional, sementara RDS dengan connection pooling (RDS Proxy) menambah biaya dan kerumitan yang tidak sepadan untuk beban sesporadis itu. DynamoDB menghindari masalah ini sepenuhnya. Setiap pemanggilan Lambda langsung terhubung tanpa batas koneksi, TTL bawaan otomatis membersihkan task yang sudah kedaluwarsa, dan kueri tetap cepat meski tabel terus membesar seiring waktu. Bagian berikut membahas desain tabel, struktur item, dan pola kueri untuk melacak penyelesaian CAPTCHA dalam arsitektur berbasis Lambda.

Desain Tabel Tunggal

Alih-alih membuat tabel terpisah untuk riwayat penyelesaian, task aktif, dan statistik, satu tabel DynamoDB menangani ketiganya sekaligus. Pola single-table ini lazim dipakai tim scraping skala menengah karena memangkas jumlah request lintas-tabel dan menyederhanakan perhitungan biaya baca/tulis.

Struktur Kunci Partisi dan Kunci Sortir

Definisi Tabel dalam JSON

Buat tabel dengan BillingMode bernilai PAY_PER_REQUEST supaya Anda tidak perlu menebak kapasitas baca/tulis sejak awal, lalu aktifkan TimeToLiveSpecification pada atribut ttl agar DynamoDB membersihkan item kedaluwarsa tanpa job terjadwal tambahan. GSI1 dipakai belakangan untuk mengueri berdasarkan status lintas semua jenis CAPTCHA.

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

Implementasi Python: Submit, Polling, dan Simpan Hasil

Pengaturan Awal

Siapkan resource DynamoDB dan API key CaptchaAI dari environment variable — pola ini memudahkan pemindahan antar stage (dev, staging, production) tanpa mengubah baris kode apa pun.

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

Fungsi solve_and_track: Kirim, Simpan, Polling, Selesai

Fungsi solve_and_track mengikuti alur empat langkah yang sama di semua integrasi CaptchaAI:

• Kirim task ke in.php
• Simpan task ID sebagai item aktif dengan TTL 10 menit
• Polling res.php setiap 5 detik sampai status bernilai 1
• Pakai token yang dikembalikan, lalu bersihkan item aktif

Setiap hasil — sukses maupun error — otomatis tersimpan sebagai record permanen dengan TTL 90 hari, item task aktif dihapus setelah selesai, dan statistik harian di partisi STATS#{date} diperbarui lewat update_daily_stats.

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
    )

Pola Kueri: Riwayat Situs, Statistik, dan Task Aktif

Tiga fungsi kueri ini menutupi kebutuhan paling umum: get_site_history mengambil solve terbaru untuk satu sitekey, get_daily_stats merangkum performa per hari, dan get_active_tasks memakai GSI1 untuk melihat task yang masih diproses. Karena semuanya memakai query, bukan scan, biayanya tetap rendah meski tabel sudah berisi jutaan item.

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

Implementasi Node.js

Versi Node.js memakai @aws-sdk/lib-dynamodb dan axios, dengan logika yang identik dengan versi Python — submit, polling tiap 5 detik, lalu simpan hasil begitu status bernilai 1. Kode ini bisa langsung dipakai sebagai handler Lambda tanpa modifikasi tambahan.

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

getSiteHistory di sini setara dengan get_site_history pada versi Python — keduanya memakai ScanIndexForward: false supaya solve terbaru selalu muncul lebih dulu.

Strategi Optimasi Biaya

DynamoDB sudah murah untuk beban kerja CAPTCHA, tapi lima kebiasaan berikut membuatnya lebih hemat lagi tanpa mengorbankan kecepatan kueri:

Catatan: gabungkan billing on-demand dengan TTL supaya biaya penyimpanan tetap minimal walau volume solve naik tajam.

Masalah Umum dan Solusinya

Empat masalah ini paling sering muncul begitu tabel mulai menerima trafik produksi:

Pertanyaan yang Sering Muncul

Kenapa data solve, task aktif, dan statistik digabung dalam satu tabel, bukan dipisah?

Pola single-table mengurangi jumlah request yang harus dilakukan Lambda per invocation — cukup satu koneksi ke satu tabel untuk membaca dan menulis semua jenis item. Untuk beban kerja CAPTCHA yang frekuensinya tinggi tapi ringan per item, ini lebih hemat daripada mengelola tiga tabel terpisah dengan billing dan izin IAM masing-masing.

Berapa perkiraan biaya DynamoDB untuk volume penyelesaian yang lebih besar?

Dengan billing on-demand, harga dasarnya sekitar $1.25 per juta penulisan dan $0.25 per juta pembacaan. Tim yang memproses 50.000 penyelesaian per hari — volume yang umum untuk agensi pemantauan harga atau operasi scraping skala menengah — biasanya masih berada di kisaran beberapa dolar per bulan untuk penyimpanan dan akses, jauh di bawah biaya operasional RDS Proxy setara.

Apakah item langsung hilang begitu TTL-nya habis?

Tidak. Penghapusan TTL di DynamoDB bersifat eventual, biasanya terjadi dalam waktu sekitar 48 jam setelah timestamp kedaluwarsa terlampaui. Jangan mengandalkan TTL untuk pembersihan real-time — filter item yang sudah kedaluwarsa langsung di level kueri kalau aplikasi Anda butuh data yang selalu bersih.

Apakah pola tabel ini bisa dipakai untuk semua jenis CAPTCHA yang didukung CaptchaAI, termasuk yang beta?

Bisa. Kolom captcha_type dan indeks GSI1 tidak bergantung pada jenis CAPTCHA tertentu, jadi pola yang sama berlaku untuk reCAPTCHA v2/v3, Cloudflare Turnstile dan Challenge, GeeTest v3, sampai CaptchaFox, Friendly Captcha, dan Lemin (ketiganya masih berstatus beta). Anda cukup menyesuaikan nilai method pada request ke in.php sesuai jenis CAPTCHA yang dikirim.

Langkah Selanjutnya

Bangun pelacakan CAPTCHA yang otomatis diskalakan tanpa server tambahan yang perlu dikelola – dapatkan API key CaptchaAI Anda.

Panduan terkait:

• Integrasi AWS Lambda dan CaptchaAI untuk CAPTCHA serverless
• Menyimpan riwayat penyelesaian CAPTCHA di MongoDB
• Mengelola TTL token CAPTCHA di Redis