ساخت API Endpoint برای دریافت Beacon از مرورگر؛ چرا یک ویو ساده کافی نیست؟ | API Endpoint برای Beacon

اگر تصور می‌کنید یک def track(request) ساده که json.loads(request.body) را اجرا کند و در دیتابیس ذخیره کند، برای دریافت beacon از مرورگر کافی است، احتمالاً تا امروز endpoint شما هم به‌عنوان دروازه‌ای برای اسپم استفاده شده و هم با گم‌شدن session، بخش بزرگی از داده‌ها را از دست داده است؛ چون beacon یک درخواست متفاوت از درخواست‌های معمولی است.

چرا beacon با درخواست معمولی متفاوت است؟

در یک درخواست معمولی HTTP، کاربر منتظر پاسخ است. اگر پاسخ دیر بیاید، کاربر ناراضی می‌شود. اگر پاسخ نیاید، کاربر متوجه می‌شود. در beacon، داستان کاملاً متفاوت است.

تفاوت اول، عدم انتظار برای پاسخ. beacon در لحظه‌ی خروج کاربر یا در لحظه‌ی یک رویداد پس‌زمینه ارسال می‌شود. کاربر منتظر پاسخ نیست. اگر پاسخ نیاید، هیچ‌کس متوجه نمی‌شود.

تفاوت دوم، عدم امکان احراز هویت. beacon نمی‌تواند هدر سفارشی بفرستد. یعنی نمی‌تواند توکن CSRF، Authorization یا هر هدر دیگری را ارسال کند.

تفاوت سوم، احتمال گم شدن. beacon ممکن است در شبکه‌های ضعیف گم شود یا توسط ad blockerها مسدود شود. برخلاف یک درخواست معمولی، شما نمی‌توانید از کاربر بخواهید دوباره ارسال کند.

تفاوت چهارم، حجم بالا. beaconها در حجم بالا تولید می‌شوند. اگر endpoint شما برای حجم معمولی طراحی شده باشد، به‌سرعت به گلوگاه تبدیل می‌شود.

این تفاوت‌ها نشان می‌دهد که طراحی endpoint beacon، یک مسئله‌ی تخصصی است، نه یک ویوی ساده. اگر با الگوی استفاده از sendBeacon برای رویداد خروج کار کرده باشید، می‌دانید که این endpoint باید در معماری کلان دیده شود.

endpoint beacon، یک دروازه‌ی یک‌طرفه است: داده از سمت کلاینت می‌آید و هیچ‌وقت پاسخ مشخصی برنمی‌گردد. طراحی این دروازه، نیازمند توجه به مسائل امنیتی، کارایی و قابلیت اعتماد است.

این اهمیت، در ساختار پروژه‌های آماری مثل اپ جنگو برای ردیابی بازدیدکننده به‌طور مستقیم دیده می‌شود. اگر endpoint beacon به‌درستی طراحی نشود، تمام داده‌های سمت کلاینت از دست می‌روند. برای درک عمیق‌تر این موضوع، پیشنهاد می‌کنم ابتدا ثبت اسکرول و کلیک کاربر را مطالعه کنید، چون endpoint beacon، مقصد نهایی این داده‌هاست.

CSRF؛ چرا beacon نمی‌تواند توکن بفرستد؟

CSRF (Cross-Site Request Forgery) یکی از رایج‌ترین حملات وب است. مکانیزم دفاعی جنگو برای این حمله، استفاده از توکن CSRF است که در فرم‌ها قرار می‌گیرد و در هر درخواست POST ارسال می‌شود. ولی beacon نمی‌تواند این توکن را بفرستد، چون:

دلیل اول، عدم امکان ارسال هدر سفارشی. beacon نمی‌تواند هدر X-CSRFToken را ارسال کند.

دلیل دوم، عدم امکان ارسال داده‌ی فرم با ساختار پیچیده. اگرچه beacon می‌تواند FormData بفرستد، ولی توکن CSRF در آن فرم قرار نمی‌گیرد.

دلیل سوم، عدم دسترسی به کوکی از سمت JavaScript. کوکی CSRF معمولاً HttpOnly نیست، ولی برای ارسال باید در هدر یا body قرار بگیرد.

به همین دلایل، endpoint beacon معمولاً با @csrf_exempt تنظیم می‌شود. ولی این کار، امنیت را پایین می‌آورد و نیازمند لایه‌های دفاعی جایگزین است.

from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST


@csrf_exempt
@require_POST
def track(request):
    ...

نکته‌ی مهم: @csrf_exempt به‌تنهایی کافی نیست. شما باید لایه‌های دفاعی جایگزین پیاده کنید.

لایه‌ی اول، بررسی Origin. هدر Origin یا Referer را چک کنید. اگر از دامنه‌ی سایت شما نبود، رد کنید.

from django.conf import settings


def check_origin(request) -> bool:
    allowed = getattr(settings, "ANALYTICS_ALLOWED_ORIGINS", [])
    if not allowed:
        return True

    origin = request.META.get("HTTP_ORIGIN", "")
    if not origin:
        # بعضی مرورگرها Origin نمی‌فرستند
        referer = request.META.get("HTTP_REFERER", "")
        if not referer:
            return False
        from urllib.parse import urlparse
        origin = f"{urlparse(referer).scheme}://{urlparse(referer).netloc}"

    return origin in allowed

لایه‌ی دوم، rate limiting. چون CSRF غیرفعال است، هر کسی می‌تواند به endpoint شما درخواست بفرستد. rate limiting، جلوی سوءاستفاده را می‌گیرد.

لایه‌ی سوم، اعتبارسنجی payload. همه‌ی فیلدها را اعتبارسنجی کنید. اندازه‌ی payload، طول فیلدها، مقادیر مجاز.

لایه‌ی چهارم، ثبت لاگ. هر درخواست مشکوک را لاگ کنید تا بتوانید حملات را تشخیص دهید.

وقتی CSRF را غیرفعال می‌کنید، مسئولیت امنیت را به عهده می‌گیرید. این مسئولیت را با لایه‌های دفاعی جایگزین، جبران کنید.

اگر با الگوی تشخیص بات از کاربر انسانی کار کرده باشید، می‌دانید که این endpoint یک نقطه‌ی حمله‌ی مهم است که باید چند لایه دفاع داشته باشد.

CORS؛ مدیریت درخواست‌های cross-origin

اگر سایت شما با دامنه‌های مختلف سرو می‌شود (مثلاً example.com و www.example.com) یا اگر از یک CDN استفاده می‌کنید که در دامنه‌ی متفاوتی است، باید CORS (Cross-Origin Resource Sharing) را مدیریت کنید.

خبر خوب این است که beacon یک درخواست "simple" است، یعنی از یک متد ساده (POST) با یک content-type ساده استفاده می‌کند. این یعنی مرورگر preflight request نمی‌فرستد و نیازی به تنظیم CORS پیچیده نیست.

ولی برای امنیت، باید در سمت سرور بررسی کنید که درخواست از یک origin معتبر آمده. این کار با هدر Origin انجام می‌شود.

def check_cors(request) -> bool:
    origin = request.META.get("HTTP_ORIGIN", "")
    if not origin:
        return True  # همان مبدأ

    allowed = getattr(settings, "ANALYTICS_ALLOWED_ORIGINS", [])
    return origin in allowed

و در settings:

ANALYTICS_ALLOWED_ORIGINS = [
    "https://example.com",
    "https://www.example.com",
]

نکته‌ی مهم: اگر از beacon روی دامنه‌های مختلف استفاده می‌کنید، باید این لیست را به‌روز نگه دارید. اگر دامنه‌ای اضافه یا حذف شود، باید این لیست را به‌روز کنید. اگر با الگوی طبقه‌بندی Referrer و تشخیص ورود از گوگل کار کرده باشید، می‌دانید که این نوع تنظیمات، باید در جای مرکزی پروژه باشد.

session؛ چالشی که اکثر پروژه‌ها را غافلگیر می‌کند

یکی از بزرگ‌ترین چالش‌ها در endpoint beacon، مدیریت session است. beacon در لحظه‌ی خروج کاربر ارسال می‌شود، ولی در آن لحظه:

چالش اول، حذف کوکی session. برخی مرورگرها، کوکی session را در لحظه‌ی بستن صفحه حذف می‌کنند. یعنی وقتی beacon به سرور می‌رسد، کوکی session وجود ندارد.

چالش دوم، عدم ارسال کوکی‌ها. beacon از همان مکانیزم ارسال کوکی مانند درخواست‌های معمولی استفاده می‌کند، ولی اگر کوکی منقضی شده باشد، ارسال نمی‌شود.

چالش سوم، مسدود شدن کوکی‌های third-party. اگر سایت شما از یک iframe یا دامنه‌ی متفاوت سرو می‌شود، کوکی‌های third-party ممکن است مسدود شوند.

برای مقابله با این چالش‌ها، سه راهبرد پیشنهاد می‌شود.

راهبرد اول، فرستادن visit_id در payload. به‌جای تکیه بر session، یک شناسه‌ی Visit در payload بفرستید.

// سمت کلاینت
const payload = {
  event: "exit",
  visit_id: window.__analytics_visit_id,
  ...
};

و در سمت سرور:

def get_visit_id(request, payload):
    # اول از session
    visit_id = request.session.get("_analytics_visit_id")
    if visit_id:
        return visit_id

    # fallback از payload
    return payload.get("visit_id")

راهبرد دوم، session در URL. یک نسخه‌ی URL-only از visit_id در کوکی غیر HttpOnly ذخیره کنید و در beacon از آن استفاده کنید.

راهبرد سوم، session بدون کوکی. اگر کاربر کوکی را مسدود کرده باشد، نمی‌توانید session را حفظ کنید. در این حالت، باید یک سشن بی‌نام بر اساس fingerprint بسازید.

def get_or_create_visit_by_fingerprint(visitor, now):
    visit = Visit.objects.filter(
        visitor=visitor,
        is_active=True,
    ).order_by("-entry_time").first()

    if visit and not _is_expired(visit, now):
        return visit

    # ساخت Visit جدید
    return Visit.objects.create(
        visitor=visitor,
        entry_time=now,
        entry_url="",
        entry_path="",
        last_activity=now,
        is_active=True,
    )

اگر با الگوی طراحی مدل Visitor و Visit در جنگو کار کرده باشید، می‌دانید که این نوع fallback، بخشی از طراحی مقاوم است.

اعتبارسنجی payload؛ خط اول دفاع

چون endpoint beacon عمومی است، هر کسی می‌تواند هر داده‌ای به آن بفرستد. اعتبارسنجی دقیق payload، خط اول دفاع است.

from django.core.validators import URLValidator
from django.core.exceptions import ValidationError


MAX_PAYLOAD_SIZE = 64 * 1024  # 64KB
MAX_EVENTS_PER_BATCH = 100
ALLOWED_EVENTS = {
    "exit", "page_hidden", "page_visible",
    "click", "scroll", "ping", "init",
}


def validate_payload(payload: dict) -> tuple[bool, str]:
    # اندازه‌ی payload
    import json
    if len(json.dumps(payload)) > MAX_PAYLOAD_SIZE:
        return False, "payload_too_large"

    # event معتبر
    if "events" in payload:
        # batching
        events = payload["events"]
        if not isinstance(events, list):
            return False, "events_not_list"
        if len(events) > MAX_EVENTS_PER_BATCH:
            return False, "too_many_events"
        for event in events:
            if not _validate_single_event(event):
                return False, "invalid_event"
    else:
        # event تک
        if not _validate_single_event(payload):
            return False, "invalid_event"

    return True, ""


def _validate_single_event(event: dict) -> bool:
    if not isinstance(event, dict):
        return False

    event_type = event.get("event", "").lower()
    if event_type not in ALLOWED_EVENTS:
        return False

    # محدودیت طول فیلدها
    for field in ("url", "exit_url", "selector", "label"):
        value = event.get(field, "")
        if isinstance(value, str) and len(value) > 2000:
            return False

    return True

این اعتبارسنجی، چند نکته‌ی کلیدی را رعایت می‌کند.

نکته‌ی اول، محدودیت اندازه‌ی کل. payload بزرگ، هم پهنای باند مصرف می‌کند، هم CPU برای پردازش JSON.

نکته‌ی دوم، whitelist رویدادها. فقط رویدادهای مجاز پردازش می‌شوند. این کار، از دیتابیس در برابر داده‌های آلوده محافظت می‌کند.

نکته‌ی سوم، محدودیت طول فیلدها. هر فیلد رشته‌ای محدودیت طول دارد. این کار، از ذخیره‌ی داده‌های بزرگ در دیتابیس جلوگیری می‌کند.

نکته‌ی مهم: این اعتبارسنجی باید قبل از هر پردازش دیگری انجام شود. اگر payload معتبر نبود، endpoint باید سریعاً پاسخ خطا بدهد، بدون مصرف منابع.

rate limiting؛ چرا بدون آن endpoint شما یک دروازه‌ی اسپم است

endpoint beacon، یک endpoint عمومی است. بدون rate limiting، هر کسی می‌تواند با ارسال هزاران درخواست در ثانیه، دیتابیس شما را پر کند یا منابع سرور را مصرف کند.

چند رویکرد برای rate limiting:

رویکرد اول، بر اساس IP. ساده‌ترین رویکرد، ولی در برابر کاربرانی که از VPN یا NAT مشترک استفاده می‌کنند، مشکل‌ساز است.

from django.core.cache import cache


def check_ip_rate_limit(ip: str, limit: int = 120, window: int = 60) -> bool:
    key = f"beacon:ip:{ip}"
    current = cache.get(key, 0)

    if current >= limit:
        return False

    if current == 0:
        cache.set(key, 1, timeout=window)
    else:
        cache.incr(key)

    return True

رویکرد دوم، بر اساس session. هر session می‌تواند حداکثر N درخواست در دقیقه بفرستد. این رویکرد، دقیق‌تر است ولی به session وابسته است.

def check_session_rate_limit(session_key: str, limit: int = 60, window: int = 60) -> bool:
    key = f"beacon:session:{session_key}"
    current = cache.get(key, 0)

    if current >= limit:
        return False

    if current == 0:
        cache.set(key, 1, timeout=window)
    else:
        cache.incr(key)

    return True

رویکرد سوم، بر اساس fingerprint. ترکیب IP و User-Agent. این رویکرد، سخت‌گیرانه‌تر از IP تنها است.

import hashlib


def check_fingerprint_rate_limit(request, limit: int = 60, window: int = 60) -> bool:
    ip = get_client_ip(request)
    ua = request.META.get("HTTP_USER_AGENT", "")
    fp = hashlib.md5(f"{ip}|{ua}".encode()).hexdigest()[:16]

    key = f"beacon:fp:{fp}"
    current = cache.get(key, 0)

    if current >= limit:
        return False

    if current == 0:
        cache.set(key, 1, timeout=window)
    else:
        cache.incr(key)

    return True

نکته‌ی مهم: باید آستانه‌ها را بر اساس رفتار واقعی کاربران کالیبره کنید. مثلاً یک کاربر عادی در یک سشن، شاید ۲۰ تا ۵۰ رویداد بفرستد. اگر آستانه را روی ۶۰ بگذارید، کاربران نرمال را مسدود نمی‌کنید، ولی بات‌ها را متوقف می‌کنید.

استراتژی	سادگی	دقت	مقاومت در برابر حمله
بر اساس IP	بالا	پایین	متوسط
بر اساس session	متوسط	بالا	بالا
بر اساس fingerprint	متوسط	بالا	بالا
ترکیب هر سه	پایین	بالا	بسیار بالا

توصیه‌ی من: در پروژه‌های جدی، از ترکیب هر سه استفاده کنید. اگر با الگوی نوشتن Middleware سفارشی در جنگو کار کرده باشید، می‌دانید که این نوع ترکیب، در پروژه‌های بزرگ استاندارد است.

idempotency؛ جلوگیری از ثبت تکراری

در شبکه‌های ناپایدار، ممکن است beacon دوبار به سرور برسد. این مشکل، در beacon‌های بزرگ‌تر (با batching) بیشتر رخ می‌دهد. برای جلوگیری از ثبت تکراری، به یک مکانیزم idempotency نیاز دارید.

ساده‌ترین راه، استفاده از یک event_id یکتا در هر رویداد و یک unique index در دیتابیس است.

class Interaction(models.Model):
    event_id = models.CharField(
        max_length=36, blank=True, db_index=True,
    )
    # ... فیلدهای دیگر

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["event_id"],
                name="unique_interaction_event",
                condition=models.Q(event_id__gt=""),
            ),
        ]

و در سرویس:

def save_interaction_safe(payload):
    event_id = payload.get("event_id")
    if event_id and Interaction.objects.filter(event_id=event_id).exists():
        return False

    Interaction.objects.create(
        event_id=event_id,
        ...
    )
    return True

نکته‌ی مهم: تولید event_id در سمت کلاینت باید یکتا باشد. یک راه، استفاده از crypto.randomUUID() در JavaScript است که UUID (Universally Unique Identifier) تولید می‌کند.

function generateEventId() {
  return crypto.randomUUID();
}

اگر با الگوی ذخیره‌ی PageView و Interaction کار کرده باشید، می‌دانید که این نوع idempotency، در سیستم‌های نوشتنی استاندارد است.

endpoint واحد در برابر endpoint دسته‌ای

دو رویکرد برای طراحی endpoint beacon وجود دارد: endpoint واحد برای هر رویداد و endpoint دسته‌ای برای گروهی از رویدادها.

ویژگی	endpoint واحد	endpoint دسته‌ای
تعداد درخواست	زیاد	کم
پیچیدگی سمت کلاینت	پایین	متوسط
پیچیدگی سمت سرور	پایین	بالا
احتمال گم شدن	بیشتر	کمتر
مناسب برای حجم بالا	خیر	بله
مناسب برای رویدادهای فوری	بله	خیر

توصیه‌ی من: در پروژه‌های جدی، از هر دو استفاده کنید.

endpoint واحد. برای رویدادهای مهم مثل exit و init. این رویدادها به‌صورت فوری ارسال می‌شوند و می‌توانند از session استفاده کنند.

endpoint دسته‌ای. برای رویدادهای پرتکرار مثل scroll و click. این رویدادها در صف جمع می‌شوند و به‌صورت گروهی ارسال می‌شوند.

# analytics/urls.py

urlpatterns = [
    path("panel/stat/api/track/", api.track_single, name="api_track"),
    path("panel/stat/api/track-batch/", api.track_batch, name="api_track_batch"),
]

و در سرویس:

# analytics/api.py

@csrf_exempt
@require_POST
def track_single(request):
    if not check_origin(request):
        return JsonResponse({"ok": False, "error": "invalid_origin"})

    if not check_ip_rate_limit(get_client_ip(request)):
        return JsonResponse({"ok": False, "error": "rate_limited"}, status=429)

    payload = _parse_json(request)
    if payload is None:
        return JsonResponse({"ok": False, "error": "invalid_json"})

    is_valid, error = validate_payload(payload)
    if not is_valid:
        return JsonResponse({"ok": False, "error": error})

    # پردازش رویداد واحد
    ...


@csrf_exempt
@require_POST
def track_batch(request):
    if not check_origin(request):
        return JsonResponse({"ok": False, "error": "invalid_origin"})

    if not check_ip_rate_limit(get_client_ip(request)):
        return JsonResponse({"ok": False, "error": "rate_limited"}, status=429)

    payload = _parse_json(request)
    if payload is None:
        return JsonResponse({"ok": False, "error": "invalid_json"})

    is_valid, error = validate_payload(payload)
    if not is_valid:
        return JsonResponse({"ok": False, "error": error})

    # پردازش دسته‌ای رویدادها
    ...

نکته‌ی مهم: هر دو endpoint باید منطق مشترک داشته باشند. برای جلوگیری از تکرار کد، منطق مشترک را در یک decorator یا یک تابع کمکی قرار دهید.

from functools import wraps


def beacon_endpoint(func):
    @wraps(func)
    @csrf_exempt
    @require_POST
    def wrapper(request, *args, **kwargs):
        if not check_origin(request):
            return JsonResponse({"ok": False, "error": "invalid_origin"})

        if not check_ip_rate_limit(get_client_ip(request)):
            return JsonResponse({"ok": False, "error": "rate_limited"}, status=429)

        payload = _parse_json(request)
        if payload is None:
            return JsonResponse({"ok": False, "error": "invalid_json"})

        is_valid, error = validate_payload(payload)
        if not is_valid:
            return JsonResponse({"ok": False, "error": error})

        return func(request, payload, *args, **kwargs)

    return wrapper


@beacon_endpoint
def track_single(request, payload):
    ...


@beacon_endpoint
def track_batch(request, payload):
    ...

این ساختار، سه مزیت دارد. اول، منطق مشترک در یک نقطه است. دوم، اگر نیاز به تغییر منطق مشترک باشد، فقط یک جا تغییر می‌کند. سوم، تست‌پذیری بالاتر است.

معماری نهایی؛ از beacon تا دیتابیس

حالا بیایید همه‌ی لایه‌ها را در یک معماری نهایی ترکیب کنیم.

لایه‌ی اول، decorator مشترک. این لایه، بررسی origin، rate limiting و اعتبارسنجی را انجام می‌دهد.

لایه‌ی دوم، دریافت رویداد. این لایه، رویداد را از payload استخراج می‌کند.

لایه‌ی سوم، سرویس. این لایه، منطق پردازش رویداد را انجام می‌دهد.

لایه‌ی چهارم، ذخیره‌سازی. این لایه، رویداد را در دیتابیس یا صف ذخیره می‌کند.

# analytics/services/beacon.py

import json
import redis
from django.conf import settings
from django.utils import timezone

from analytics.models import Visit, PageView, Interaction


redis_client = redis.Redis.from_url(settings.REDIS_URL)


def process_single_event(request, payload):
    visit_id = _resolve_visit_id(request, payload)
    if not visit_id:
        return {"ok": False, "error": "no_visit"}

    event_type = payload.get("event", "").lower()

    if event_type == "exit":
        return _process_exit(visit_id, payload)

    if event_type in ("page_hidden", "page_visible"):
        return _process_visibility(visit_id, payload)

    if event_type == "ping":
        return _process_ping(visit_id, payload)

    return {"ok": False, "error": "unknown_event"}


def process_batch_events(request, payload):
    visit_id = _resolve_visit_id(request, payload)
    if not visit_id:
        return {"ok": False, "error": "no_visit"}

    events = payload.get("events", [])
    if not events:
        return {"ok": False, "error": "no_events"}

    # به صف Redis اضافه کن
    batch = {
        "visit_id": visit_id,
        "page_view_id": request.session.get("_analytics_page_view_id"),
        "events": events,
        "received_at": timezone.now().isoformat(),
    }

    try:
        redis_client.lpush(
            "analytics:interactions",
            json.dumps(batch, ensure_ascii=False),
        )
    except Exception as exc:
        return {"ok": False, "error": "queue_failed"}

    return {"ok": True, "count": len(events)}


def _resolve_visit_id(request, payload):
    visit_id = request.session.get("_analytics_visit_id")
    if visit_id:
        return visit_id
    return payload.get("visit_id")


def _process_exit(visit_id, payload):
    from analytics.services.exit import close_visit
    success = close_visit(visit_id, payload)
    return {"ok": success}


def _process_visibility(visit_id, payload):
    from analytics.services.exit import touch_visit, mark_visit_active

    event_type = payload.get("event", "").lower()
    if event_type == "page_hidden":
        touch_visit(visit_id)
    else:
        mark_visit_active(visit_id)

    return {"ok": True}


def _process_ping(visit_id, payload):
    from analytics.services.exit import touch_visit
    touch_visit(visit_id)
    return {"ok": True}

این معماری، سه مزیت کلیدی دارد. اول، منطق در یک سرویس متمرکز است. دوم، endpoint فقط نقش هماهنگی دارد. سوم، صف Redis، مقاومت در برابر اسپایک را فراهم می‌کند.

اگر با الگوی تعریف و کاربرد services.py در جنگو کار کرده باشید، این جداسازی برایتان آشناست.

صف‌بندی با Redis؛ جایی که مقیاس‌پذیری شروع می‌شود

در پروژه‌های با ترافیک بالا، نوشتن مستقیم در دیتابیس می‌تواند گلوگاه شود. راه‌حل، استفاده از یک صف است.

سه مزیت اصلی صف:

مزیت اول، پاسخ فوری. endpoint بلافاصله پاسخ می‌دهد، بدون انتظار برای دیتابیس.

مزیت دوم، مقاومت در برابر اسپایک. اگر ترافیک ناگهان چند برابر شود، صف می‌تواند حجم بالایی را تحمل کند.

مزیت سوم، انعطاف‌پذیری. اگر منطق پردازش تغییر کند، فقط worker را تغییر می‌دهیم، نه endpoint را.

پیاده‌سازی صف با Redis ساده است:

import redis
from django.conf import settings


redis_client = redis.Redis.from_url(settings.REDIS_URL, decode_responses=False)


def enqueue_event(queue_name: str, payload: dict):
    import json
    redis_client.lpush(
        queue_name,
        json.dumps(payload, ensure_ascii=False).encode("utf-8"),
    )


def dequeue_events(queue_name: str, batch_size: int = 100) -> list:
    import json
    items = []
    for _ in range(batch_size):
        item = redis_client.rpop(queue_name)
        if item is None:
            break
        try:
            items.append(json.loads(item))
        except Exception:
            continue
    return items

و در یک management command یا Celery task:

# analytics/management/commands/process_beacons.py

import time
from django.core.management.base import BaseCommand

from analytics.services.beacon import dequeue_events
from analytics.services.interaction import save_interactions_batch


class Command(BaseCommand):
    help = "پردازش صف beaconها"

    def handle(self, *args, **options):
        while True:
            batch = dequeue_events("analytics:interactions", batch_size=100)

            if not batch:
                time.sleep(1)
                continue

            for item in batch:
                try:
                    save_interactions_batch(
                        visit_id=item["visit_id"],
                        page_view_id=item.get("page_view_id"),
                        events=item["events"],
                    )
                except Exception as exc:
                    self.stderr.write(f"Error: {exc}")

نکته‌ی مهم: در production، از یک process manager مثل Supervisor یا systemd برای اجرای worker استفاده کنید. اگر با الگوی بهینه‌سازی جنگو برای ترافیک بالا کار کرده باشید، می‌دانید که این نوع معماری، در پروژه‌های بزرگ استاندارد است.

امنیت؛ هفت لایه‌ی محافظت

endpoint beacon، به‌خاطر عمومی بودن، سطح حمله‌ی بزرگی دارد. هفت لایه‌ی محافظت را در نظر بگیرید.

لایه‌ی اول، بررسی Origin. همان‌طور که در بخش CSRF توضیح دادم.

لایه‌ی دوم، rate limiting. بر اساس IP، session و fingerprint.

لایه‌ی سوم، اعتبارسنجی payload. طول، نوع، مقادیر مجاز.

لایه‌ی چهارم، idempotency. جلوگیری از ثبت تکراری.

لایه‌ی پنجم، محدودیت طول فیلدها. هر فیلد رشته‌ای، محدودیت طول دارد.

لایه‌ی ششم، لاگ‌گیری. هر درخواست مشکوک را لاگ کنید.

لایه‌ی هفتم، نظارت. اگر نرخ درخواست‌ها به‌طور ناگهانی افزایش یافت، هشدار دریافت کنید.

import logging

logger = logging.getLogger("analytics.beacon")


def log_suspicious_request(request, reason: str):
    logger.warning(
        "Suspicious beacon request",
        extra={
            "reason": reason,
            "ip": get_client_ip(request),
            "user_agent": request.META.get("HTTP_USER_AGENT", "")[:200],
            "origin": request.META.get("HTTP_ORIGIN", ""),
            "path": request.path,
        },
    )

نکته‌ی مهم: لاگ‌ها را در یک سیستم متمرکز (Sentry، ELK، Loki) جمع کنید. بدون این، لاگ‌ها در سکوت اتفاق می‌افتند.

endpoint beacon، به‌خاطر عمومی بودن، باید مثل یک endpoint احراز هویت محافظت شود. حتی اگر داده‌ای که می‌فرستد حساس نیست، خود endpoint یک سطح حمله است.

تست‌نویسی endpoint beacon

سه سطح تست را در نظر بگیرید.

سطح اول، تست اعتبارسنجی. payloadهای مختلف را تست کنید:

import pytest
from analytics.services.beacon import validate_payload


@pytest.mark.parametrize("payload,valid", [
    ({"event": "ping"}, True),
    ({"event": "unknown"}, False),
    ({"event": "click", "label": "x" * 3000}, False),
    ({"events": []}, False),
    ({"events": [{"event": "scroll"} for _ in range(150)]}, False),
    ({"events": [{"event": "scroll"} for _ in range(50)]}, True),
])
def test_validate_payload(payload, valid):
    is_valid, error = validate_payload(payload)
    assert is_valid == valid

سطح دوم، تست rate limiting. رفتار rate limiting را تست کنید:

@pytest.mark.django_db
def test_rate_limit(client):
    from django.core.cache import cache
    cache.clear()

    for i in range(100):
        response = client.post(
            "/panel/stat/api/track/",
            data=json.dumps({"event": "ping"}),
            content_type="application/json",
            HTTP_ORIGIN="https://example.com",
        )

    # بعد از حد مجاز، پاسخ باید 429 باشد
    assert response.status_code == 429

سطح سوم، تست idempotency. بررسی کنید که ارسال دوباره‌ی یک event_id باعث ثبت دوباره نمی‌شود:

@pytest.mark.django_db
def test_idempotency(client):
    from analytics.models import Interaction

    payload = {
        "event": "click",
        "event_id": "test-uuid-1234",
        "selector": "#buy",
    }

    client.post(
        "/panel/stat/api/track/",
        data=json.dumps(payload),
        content_type="application/json",
    )
    client.post(
        "/panel/stat/api/track/",
        data=json.dumps(payload),
        content_type="application/json",
    )

    assert Interaction.objects.filter(event_id="test-uuid-1234").count() == 1

این سه سطح تست، به شما اجازه می‌دهند که در طول زمان، تغییرات را با اطمینان اعمال کنید.

anti-patternهای رایج در طراحی endpoint beacon

در بازبینی پروژه‌های مختلف، این اشتباهات را زیاد دیده‌ام:

۱. عدم بررسی Origin. بدون این بررسی، هر سایتی می‌تواند به endpoint شما درخواست بفرستد.

۲. عدم rate limiting. بدون این لایه، endpoint شما یک دروازه‌ی اسپم است.

۳. عدم اعتبارسنجی payload. هر داده‌ای در دیتابیس ذخیره می‌شود، شامل داده‌های آلوده.

۴. تکیه بر session در beacon خروج. در لحظه‌ی خروج، ممکن است session در دسترس نباشد.

۵. نوشتن مستقیم در دیتابیس. در ترافیک بالا، این کار به گلوگاه تبدیل می‌شود.

۶. عدم idempotency. در شبکه‌های ناپایدار، رویدادها دوبار ثبت می‌شوند.

۷. عدم محدودیت طول فیلدها. payload بزرگ، دیتابیس را پر می‌کند.

۸. عدم لاگ‌گیری درخواست‌های مشکوک. بدون لاگ، نمی‌توانید حملات را تشخیص دهید.

۹. endpoint واحد برای همه‌ی رویدادها. بدون تفکیک رویدادهای فوری از پرتکرار، endpoint همیشه سنگین می‌ماند.

۱۰. عدم صف‌بندی در پروژه‌های بزرگ. بدون صف، در اسپایک ترافیک، endpoint از کار می‌افتد.

۱۱. عدم پشتیبانی از batching. بدون batching، تعداد درخواست‌های HTTP بالا می‌رود.

۱۲. عدم تست کارایی. بدون تست کارایی، رگرسیون‌های پنهان در production مشکل‌ساز می‌شوند.

اگر با الگوی تشخیص بات از کاربر انسانی کار کرده باشید، می‌دانید که این نوع لایه‌بندی، در پروژه‌های بزرگ ضروری است.

پرسش‌های پرتکرار درباره‌ی API Beacon

آیا باید endpoint beacon را با CSRF محافظت کنم؟ beacon نمی‌تواند توکن CSRF بفرستد، پس باید @csrf_exempt بگذارید. ولی به‌جای آن، باید لایه‌های دفاعی جایگزین پیاده کنید.

آیا باید از HTTPS استفاده کنم؟ بله، همیشه. beacon روی HTTP در مرورگرهای مدرن ممکن است مسدود شود و داده‌ی شما گم شود.

چطور با ad blockerها مقابله کنم؟ ad blockerها اغلب درخواست‌های به مسیرهای خاص مثل /analytics/ یا /track/ را مسدود می‌کنند. توصیه می‌کنم از یک مسیر عمومی مثل /api/events/ استفاده کنید.

چطور با کوکی‌های third-party مسدودشده مقابله کنم؟ session را از URL یا از fingerprint بسازید.

آیا باید در beacon، هدر Authorization بفرستم؟ beacon نمی‌تواند هدر سفارشی بفرستد. برای احراز هویت، از درخواست‌های معمولی استفاده کنید.

چطور اندازه‌ی payload را کاهش دهم؟ فقط داده‌های ضروری را بفرستید، از فرمت‌های سبک استفاده کنید، از batching استفاده کنید.

آیا باید برای هر رویداد یک endpoint جدا داشته باشم؟ نه، یک endpoint با پارامتر event کافی است. endpoint‌های جداگانه، نگهداری را پیچیده می‌کنند.

چطور با حملات DDoS (Distributed Denial of Service) مقابله کنم؟ از CDN و WAF (Web Application Firewall) استفاده کنید. اگر با الگوی بهینه‌سازی جنگو برای ترافیک بالا کار کرده باشید، می‌دانید که این لایه، در پروژه‌های بزرگ ضروری است.

آیا باید beacon را در فایل‌های log ذخیره کنم؟ برای ذخیره‌ی طولانی‌مدت، بله. برای تحلیل سریع، دیتابیس. اگر می‌خواهید معماری کامل را ببینید، ذخیره‌ی PageView و Interaction راهنمای دقیقی است.

چطور با کاربرانی که از VPN استفاده می‌کنند مقابله کنم؟ VPN روی beacon تأثیر نمی‌گذارد، چون beacon در سطح HTTP ارسال می‌شود. IP در سمت سرور، IP VPN خواهد بود.

آیا باید beacon را در دیتابیس جداگانه ذخیره کنم؟ در پروژه‌های بزرگ، بله. با database router می‌توانید write و read را جدا کنید.

چطور از idempotency در batching استفاده کنم؟ به هر رویداد یک event_id یکتا اختصاص دهید و در دیتابیس یک unique index بگذارید.

آیا باید endpoint beacon را در robots.txt مسدود کنم؟ بله، این یک اقدام احتیاطی است. ربات‌ها نباید تلاش کنند که این مسیر را ایندکس کنند.

چطور با مرورگرهای قدیمی که sendBeacon را پشتیبانی نمی‌کنند، مقابله کنم؟ از fetch با keepalive به‌عنوان fallback استفاده کنید.

آیا باید endpoint beacon را با autentication محافظت کنم؟ نه، beacon نمی‌تواند احراز هویت کند. ولی می‌توانید از یک توکن در payload استفاده کنید که در سمت سرور اعتبارسنجی شود.

نگاهی از منظر مهندس داده در مقیاس میلیونی

در مقیاس میلیون‌ها beacon در روز، endpoint beacon تبدیل به یک مسئله‌ی مهندسی داده می‌شود. سه مفهوم بنیادین را باید بازتعریف کنید.

مفهوم اول، back-pressure. اگر صف Redis پر شود، endpoint باید back-pressure اعمال کند. این کار با پاسخ ۴۲۹ (Too Many Requests) به بعضی درخواست‌ها انجام می‌شود. بدون back-pressure، صف پر می‌شود و endpoint از کار می‌افتد.

def check_queue_capacity(queue_name: str, max_size: int = 100000) -> bool:
    size = redis_client.llen(queue_name)
    return size < max_size

مفهوم دوم، priority queue. نه همه‌ی رویدادها به یک اندازه مهم هستند. رویدادهای مهم مثل exit و purchase باید در اولویت بالاتری قرار بگیرند. این کار با استفاده از چند صف جداگانه انجام می‌شود.

HIGH_PRIORITY_EVENTS = {"exit", "purchase", "form_submit"}
NORMAL_PRIORITY_EVENTS = {"click", "scroll", "page_hidden"}


def enqueue_event(event_type: str, payload: dict):
    if event_type in HIGH_PRIORITY_EVENTS:
        queue_name = "analytics:interactions:high"
    else:
        queue_name = "analytics:interactions:normal"

    redis_client.lpush(queue_name, json.dumps(payload).encode("utf-8"))

مفهوم سوم، monitoring. در مقیاس بالا، باید به‌طور مداوم اندازه‌ی صف، نرخ درخواست و زمان پاسخ endpoint را مانیتور کنید. اگر یکی از این متریک‌ها به آستانه رسید، باید هشدار دریافت کنید.

نکته‌ی آخر: در مقیاس بالا، هیچ endpoint کاملی وجود ندارد. هدف، رسیدن به یک trade-off قابل قبول بین دقت و کارایی است. بهترین کاری که می‌توانید بکنید این است که این trade-off را صریح کنید و به‌طور مداوم آن را اندازه بگیرید. اگر با الگوی ساخت API JSON برای آمار زنده کار کرده باشید، می‌دانید که این نوع اندازه‌گیری مداوم، بخشی از انضباط مهندسی داده است.

پرسشی که در پایان باید پاسخ دهید

قبل از اینکه endpoint beacon خود را نهایی کنید، یک پرسش را از خودتان بپرسید: «اگر امروز یک مهاجم با یک اسکریپت ساده، هزار درخواست در ثانیه به این endpoint بفرستد، سیستم من چطور واکنش نشان می‌دهد؟» اگر پاسخ شما «منابع سرور مصرف می‌شود» یا «دیتابیس پر می‌شود» است، یعنی سیستم شما به لایه‌های دفاعی بیشتری نیاز دارد. اگر پاسخ شما «بخشی از درخواست‌ها با ۴۲۹ رد می‌شوند، صف بی‌محابا پر نمی‌شود و سایت پایدار می‌ماند» است، یعنی سیستم شما به‌درستی طراحی شده است.

طراحی endpoint beacon، در نهایت یک تصمیم مهندسی است که به کیفیت داده و پایداری سیستم شما گره خورده است. اگر این تجربه را در پروژه‌ی خودتان داشته‌اید — مثلاً جایی که endpoint شما هدف حمله قرار گرفته یا جایی که session کاربران گم شده — برایم جالب است بدانید. مخصوصاً اگر راه‌حل خاصی برای یک سناریوی خاص پیدا کرده‌اید، چون همان راه‌حل‌ها می‌توانند به خواننده‌ی بعدی کمک کنند. تجربه‌ی خودتان را در دیدگاه‌ها بنویسید.