Cart Abandonment Beacon Server (pseudo-code) - heartbeat/visibility handler that schedules abandonment checks on silence.

cart_abandonment_beacon_server.py

"""
Cart Abandonment - Beacon Server (Pseudo-code)

Server-side handler for the beacon client
(https://gist.github.com/ceaksan/cart_abandonment_beacon_client.js).

Maintains per-session state and schedules an abandonment check after
each heartbeat. If no heartbeat arrives within the grace window
(e.g. 30 minutes) and no purchase has been recorded, an abandonment
is triggered.

Session store can be Redis, Cloudflare KV, Firestore or anything with
TTL support. Scheduling can be a cron job scanning for expired sessions
or a task queue with delayed execution.

See: https://ceaksan.com/tr/sepet-terk-takibi-javascript
License: MIT
"""

from datetime import datetime, timedelta


def handle_beacon(payload: dict) -> None:
    """POST /cart-beacon handler."""
    session_id = payload["sessionId"]
    event_type = payload["eventType"]

    if event_type == "purchase_complete":
        # Purchase landed, cancel any scheduled abandonment
        cancel_abandonment_check(session_id)
        delete_session(session_id)
        return

    if event_type in ("page_enter", "heartbeat", "tab_visible"):
        # User is active, refresh session
        update_session(
            session_id,
            {
                "last_ping": now(),
                "cart_value": payload.get("cartValue"),
                "url": payload["url"],
            },
        )
        # Re-schedule the abandonment check 30 minutes out
        schedule_abandonment_check(session_id, delay_minutes=30)
        return

    if event_type == "tab_hidden":
        # Tab hidden, potential abandonment
        update_session(session_id, {"last_ping": now(), "tab_hidden": True})
        schedule_abandonment_check(session_id, delay_minutes=30)


def scheduled_abandonment_check(session_id: str) -> None:
    """Runs when the abandonment timer fires for a session."""
    session = get_session(session_id)
    if not session:
        return

    # If no heartbeat in the last 25 minutes and no purchase, trigger abandonment
    if now() - session["last_ping"] > timedelta(minutes=25):
        trigger_abandonment(
            session_id=session_id,
            cart_value=session["cart_value"],
            last_url=session["url"],
            time_spent=session["last_ping"] - session["created_at"],
        )
        delete_session(session_id)


# --- Infrastructure shims (implement per your stack) ---


def now() -> datetime:
    return datetime.utcnow()


def get_session(session_id: str) -> dict | None: ...
def update_session(session_id: str, patch: dict) -> None: ...
def delete_session(session_id: str) -> None: ...
def schedule_abandonment_check(session_id: str, delay_minutes: int) -> None: ...
def cancel_abandonment_check(session_id: str) -> None: ...
def trigger_abandonment(**kwargs) -> None: ...