Giải CAPTCHA theo sự kiện với AWS SNS và CaptchaAI

Khi một kết quả CAPTCHA cần đến nhiều nơi cùng lúc — hàng đợi xử lý, log kiểm tra, cảnh báo lỗi — thì để mỗi service tự polling res.php là cách tốn kém nhất. Gọn hơn: CaptchaAI gửi kết quả về một callback, callback đẩy message lên AWS SNS topic, rồi mọi consumer phía sau (SQS, Lambda, email) tự nhận và xử lý độc lập. Bài này dựng mô hình fan-out theo sự kiện đó qua năm bước, kèm code Python và Node.js chạy được ngay.

Kiến trúc: một kết quả, nhiều consumer

Ý tưởng cốt lõi là tách trình nhận callback khỏi phần xử lý kết quả. Callback chỉ làm một việc: đẩy kết quả lên SNS topic rồi trả 200 ngay.

[Scraper] → Submit CAPTCHA → [CaptchaAI API]
                                    ↓
                            Solve completes
                                    ↓
                            Callback → [API Gateway + Lambda]
                                    ↓
                            Publish → [SNS Topic]
                                    ↓
                    ┌───────────────┼───────────────┐
                    ↓               ↓               ↓
            [SQS Queue]      [Lambda Logger]   [Email Alert]
            (result store)   (audit trail)     (on failure)

SNS lo phần fan-out: một kết quả giải CAPTCHA kích hoạt nhiều consumer mà callback không cần biết chúng tồn tại — muốn thêm service phân tích mới thì chỉ subscribe thêm vào topic. Mô hình này rất hợp với một team QA gia công ở TP.HCM chạy kiểm thử trên hàng chục cổng khách hàng: cùng một kết quả vừa vào hàng đợi xử lý, vừa ghi log đối soát, vừa bắn cảnh báo cho người trực — ba nhánh chạy song song, không nhánh nào chờ nhánh nào.

Bước 1: Tạo SNS topic

Topic là điểm phát trung tâm. Tạo một lần bằng AWS CLI (rồi lưu lại TopicArn để dùng ở các bước sau), hoặc làm tương tự bằng boto3 trong Python:

aws sns create-topic --name captcha-results --output text
# Returns: arn:aws:sns:us-east-1:123456789:captcha-results

import boto3

sns = boto3.client("sns", region_name="us-east-1")

response = sns.create_topic(Name="captcha-results")
topic_arn = response["TopicArn"]
print(f"Topic ARN: {topic_arn}")

Bước 2: Dựng Lambda nhận callback

Hàm Lambda dưới đây nhận callback từ CaptchaAI qua API Gateway, đóng gói thành message rồi publish lên SNS. Nó giữ mỏng — chỉ chuyển tiếp — để trả 200 thật nhanh, tránh timeout phía CaptchaAI. Dưới đây là bản Python và bản Node.js tương đương:

import json
import os
import boto3

sns = boto3.client("sns")
TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]

def lambda_handler(event, context):
    """Receive CaptchaAI callback and publish to SNS."""
    # Parse query parameters from API Gateway
    params = event.get("queryStringParameters", {}) or {}
    task_id = params.get("id", "")
    solution = params.get("code", "")

    if not task_id or not solution:
        return {"statusCode": 400, "body": "Missing id or code"}

    # Publish to SNS
    message = {
        "task_id": task_id,
        "solution": solution,
        "status": "solved"
    }

    sns.publish(
        TopicArn=TOPIC_ARN,
        Message=json.dumps(message),
        Subject="captcha-solved",
        MessageAttributes={
            "task_id": {
                "DataType": "String",
                "StringValue": task_id
            }
        }
    )

    return {"statusCode": 200, "body": "OK"}

const { SNSClient, PublishCommand } = require("@aws-sdk/client-sns");

const sns = new SNSClient({ region: "us-east-1" });
const TOPIC_ARN = process.env.SNS_TOPIC_ARN;

exports.handler = async (event) => {
  const params = event.queryStringParameters || {};
  const taskId = params.id;
  const solution = params.code;

  if (!taskId || !solution) {
    return { statusCode: 400, body: "Missing id or code" };
  }

  const message = {
    task_id: taskId,
    solution: solution,
    status: "solved",
  };

  await sns.send(
    new PublishCommand({
      TopicArn: TOPIC_ARN,
      Message: JSON.stringify(message),
      Subject: "captcha-solved",
      MessageAttributes: {
        task_id: { DataType: "String", StringValue: taskId },
      },
    })
  );

  return { statusCode: 200, body: "OK" };
};

Bước 3: Gửi CAPTCHA kèm URL callback

Khi gửi task, trỏ tham số pingback vào endpoint API Gateway ở bước 2. Giải xong, CaptchaAI tự gọi URL này kèm id và code, không cần polling res.php:

import os
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
CALLBACK_URL = os.environ["CALLBACK_GATEWAY_URL"]  # API Gateway URL

def submit_captcha(sitekey, pageurl):
    """Submit CAPTCHA with SNS-backed callback."""
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "pingback": CALLBACK_URL,
        "json": 1
    })
    data = resp.json()

    if data.get("status") == 1:
        return data["request"]  # task_id
    raise RuntimeError(f"Submit failed: {data.get('request')}")

Bước 4: Đăng ký consumer vào topic

Mỗi kiểu xử lý là một subscription riêng: SQS lưu kết quả cho scraper đọc, Lambda ghi log kiểm tra, và email cảnh báo khi có lỗi. Ba đoạn dưới đây đăng ký lần lượt từng loại:

# Subscribe an SQS queue to receive all results
sqs_arn = "arn:aws:sqs:us-east-1:123456789:captcha-results-queue"

sns.subscribe(
    TopicArn=topic_arn,
    Protocol="sqs",
    Endpoint=sqs_arn
)

# Subscribe a Lambda for audit logging
lambda_arn = "arn:aws:lambda:us-east-1:123456789:function:captcha-audit-logger"

sns.subscribe(
    TopicArn=topic_arn,
    Protocol="lambda",
    Endpoint=lambda_arn
)

# Subscribe email for error notifications with filter
sns.subscribe(
    TopicArn=topic_arn,
    Protocol="email",
    Endpoint="ops@example.com"
)

Bước 5: Đọc kết quả từ SQS

Giờ scraper đọc thẳng từ SQS thay vì hỏi CaptchaAI. Dùng long polling (tối đa 20 giây) để giảm request rỗng, và xóa message sau khi xử lý xong — bản Python và Node.js:

import json
import boto3

sqs = boto3.client("sqs", region_name="us-east-1")
QUEUE_URL = os.environ["SQS_QUEUE_URL"]

def get_solved_captcha(timeout=30):
    """Wait for a CAPTCHA solution from the SQS queue."""
    response = sqs.receive_message(
        QueueUrl=QUEUE_URL,
        MaxNumberOfMessages=1,
        WaitTimeSeconds=min(timeout, 20)  # Long polling (max 20s)
    )

    messages = response.get("Messages", [])
    if not messages:
        return None

    msg = messages[0]
    # SNS wraps the message — unwrap it
    sns_envelope = json.loads(msg["Body"])
    result = json.loads(sns_envelope["Message"])

    # Delete message after processing
    sqs.delete_message(
        QueueUrl=QUEUE_URL,
        ReceiptHandle=msg["ReceiptHandle"]
    )

    return result

const {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
} = require("@aws-sdk/client-sqs");

const sqs = new SQSClient({ region: "us-east-1" });
const QUEUE_URL = process.env.SQS_QUEUE_URL;

async function getSolvedCaptcha(timeout = 30) {
  const response = await sqs.send(
    new ReceiveMessageCommand({
      QueueUrl: QUEUE_URL,
      MaxNumberOfMessages: 1,
      WaitTimeSeconds: Math.min(timeout, 20),
    })
  );

  const messages = response.Messages || [];
  if (messages.length === 0) return null;

  const msg = messages[0];
  const snsEnvelope = JSON.parse(msg.Body);
  const result = JSON.parse(snsEnvelope.Message);

  await sqs.send(
    new DeleteMessageCommand({
      QueueUrl: QUEUE_URL,
      ReceiptHandle: msg.ReceiptHandle,
    })
  );

  return result;
}

Lọc message ngay trong SNS

Không phải consumer nào cũng cần mọi kết quả. Với filter policy, bạn định tuyến theo thuộc tính message — ví dụ chỉ đẩy ca thất bại sang hàng đợi vận hành:

# Only send failures to the ops queue
sns.subscribe(
    TopicArn=topic_arn,
    Protocol="sqs",
    Endpoint=failure_queue_arn,
    Attributes={
        "FilterPolicy": json.dumps({
            "status": ["failed", "error"]
        })
    }
)

Câu hỏi thường gặp

SNS có thay thế hoàn toàn việc polling res.php không?

Có, một khi callback chạy ổn thì scraper đọc từ SQS thay vì polling. Nên giữ polling làm phương án dự phòng khi callback hoặc API Gateway gặp sự cố.

Thêm nhiều consumer qua SNS có làm tăng phí giải CAPTCHA không?

Không. CaptchaAI tính theo thread (số luồng giải đồng thời), không theo số lần giải, nên gắn thêm consumer không làm tăng phí giải — bạn vẫn trả theo gói thread, ví dụ BASIC ($15/tháng, 5 thread). Chi phí SNS và SQS do AWS tính riêng.

Nếu một consumer xử lý lỗi thì kết quả có bị mất không?

Không, nếu consumer là SQS: message nằm trong hàng đợi cho đến khi được xóa thành công. Cấu hình thêm dead-letter queue để giữ lại các message thất bại, tránh mất dữ liệu.

Xử lý sự cố thường gặp

Bài viết liên quan

• Xây dựng pipeline CAPTCHA cho khách hàng
• Tự động hóa có trách nhiệm với CaptchaAI
• Giám sát mức dùng CaptchaAI qua dashboard

Bước tiếp theo

Dựng pipeline giải CAPTCHA theo sự kiện — lấy API key CaptchaAI và nối vào luồng sự kiện AWS.

Hướng dẫn liên quan:

• Tích hợp AWS Lambda serverless với CaptchaAI
• Hướng dẫn URL callback và webhook
• Bảo mật webhook: xác thực callback