keepalive.py

"""
keepalive.py — server-side "Keep Awake" for B24 Browser.

SCOPE (read this before wiring it up):
This pings a URL periodically from the backend, so a tracked tab stays
"active" even while the user's phone is asleep or the app is backgrounded
or fully closed — as long as this backend itself keeps running (note: a
free HF Space can also idle-sleep, so this isn't infinite either).

What this does NOT do: it does not simulate mouse movement, keystrokes,
or run the page's real JavaScript. It's a plain periodic HTTP request.
That's enough for many simple session-timeout mechanisms (dashboards,
admin panels, some web tools) but will NOT fool sites that check real
interaction via WebSocket heartbeats or JS-side activity checks — those
need a full headless browser per session, which isn't safe to run at
scale on limited RAM, and many such platforms (e.g. hosted notebook /
compute services) explicitly prohibit keep-alive workarounds in their
ToS. Keep this scoped to "don't lose my session on a page I'm using
while I multitask", not "trick any site indefinitely" — and maybe warn
users in the UI if a URL looks like a known compute/notebook host.
"""

import time
import threading
import requests

MAX_SESSIONS_PER_USER = 3
MAX_DURATION_SECONDS = 6 * 60 * 60  # 6h safety cap, then auto-stop
MIN_INTERVAL_SECONDS = 30

_sessions = {}  # session_id -> dict(user_id, url, interval, started_at, stop_event, thread)
_lock = threading.Lock()


def _run(session_id, url, interval, headers, stop_event, started_at):
    while not stop_event.is_set():
        if time.time() - started_at > MAX_DURATION_SECONDS:
            break
        try:
            requests.get(url, headers=headers or {}, timeout=10)
        except requests.RequestException:
            pass  # transient failure — just try again next tick
        stop_event.wait(interval)

    with _lock:
        _sessions.pop(session_id, None)


def start(user_id: str, url: str, interval: int = 60, headers: dict = None):
    """Begin pinging `url` every `interval` seconds on behalf of user_id."""
    interval = max(interval, MIN_INTERVAL_SECONDS)

    with _lock:
        active_for_user = [s for s in _sessions.values() if s["user_id"] == user_id]
        if len(active_for_user) >= MAX_SESSIONS_PER_USER:
            return {"error": f"max {MAX_SESSIONS_PER_USER} keep-alive sessions per user"}

        session_id = f"{user_id}:{int(time.time() * 1000)}"
        stop_event = threading.Event()
        thread = threading.Thread(
            target=_run,
            args=(session_id, url, interval, headers, stop_event, time.time()),
            daemon=True,
        )
        _sessions[session_id] = {
            "user_id": user_id,
            "url": url,
            "interval": interval,
            "started_at": time.time(),
            "stop_event": stop_event,
            "thread": thread,
        }
        thread.start()

    return {"session_id": session_id, "interval": interval, "max_duration": MAX_DURATION_SECONDS}


def stop(user_id: str, session_id: str):
    with _lock:
        session = _sessions.get(session_id)
        if not session or session["user_id"] != user_id:
            return {"error": "session not found"}
        session["stop_event"].set()
    return {"stopped": True}


def status(user_id: str):
    with _lock:
        return [
            {
                "session_id": sid,
                "url": s["url"],
                "interval": s["interval"],
                "elapsed_seconds": int(time.time() - s["started_at"]),
                "remaining_seconds": max(0, MAX_DURATION_SECONDS - int(time.time() - s["started_at"])),
            }
            for sid, s in _sessions.items()
            if s["user_id"] == user_id
        ]