creator-hub/browser_gateway/server/http.py

"""CreatorHub native browser gateway."""

from __future__ import annotations

import hmac
import json
import logging
import math
import os
import re
import secrets
import signal
import socket
import threading
import time
from collections.abc import Mapping
from contextlib import suppress
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import cast
from urllib.parse import parse_qs, urlsplit

from ..platform.douyin import (
    ACCOUNT_KEY_RE,
    ACTIONS,
    COMMENTS_PATH,
    IDENTITY_URL,
    UID_RE,
    WORKS_PATH,
    DouyinBrowser,
    DouyinError,
    SubscriptionManager,
    is_douyin_content_url,
    is_douyin_share_url,
)
from ..proxy import ProxyExit
from ..runtime import (
    NETWORK_ID_RE as _NETWORK_ID_RE,
)
from ..runtime import (
    RUNTIME_CLEANUP_SENTINEL as _RUNTIME_CLEANUP_SENTINEL,
)
from ..runtime import (
    RUNTIME_ID_RE as _RUNTIME_ID_RE,
)
from ..runtime import (
    BrowserRuntimeError,
    NativeRuntimeManager,
    has_control,
    validate_runtime_input,
)
from ..runtime import (
    parse_proxy_exit as _parse_proxy_exit,
)
from ..runtime import (
    validate_proxy_exit as _validate_proxy_exit,
)

LOG = logging.getLogger("creatorhub.gateway")
RUNTIME_ID_RE = _RUNTIME_ID_RE
NETWORK_ID_RE = _NETWORK_ID_RE
RUNTIME_CLEANUP_SENTINEL = _RUNTIME_CLEANUP_SENTINEL
EXIT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$")
DOUYIN_ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,127}$")
DOUYIN_ORIGIN = "https://www.douyin.com"
DOUYIN_IDENTITY_PATH = "/aweme/v1/web/user/profile/self/"
DOUYIN_PROFILE_OTHER_PATH = "/aweme/v1/web/user/profile/other/"
DOUYIN_IDENTITY_URL = IDENTITY_URL
DOUYIN_WORKS_PATH = WORKS_PATH
DOUYIN_WORK_DETAIL_PATH = "/aweme/v1/web/aweme/detail/"
DOUYIN_COMMENTS_PATH = COMMENTS_PATH


def _noop() -> None:
    return None


class RequestError(BrowserRuntimeError):
    pass


class Gateway:
    def __init__(
        self,
        runtimes: NativeRuntimeManager,
        token: str,
        node_id: str,
        browser: DouyinBrowser | None = None,
    ) -> None:
        self.runtimes = runtimes
        self.token = token
        self.node_id = node_id
        self.browser = browser or DouyinBrowser(self._browser_endpoint)
        self.proxies = runtimes.proxies
        self.subscriptions = SubscriptionManager(self.browser)
        self._action_ownership_lock = threading.Lock()
        self._uncertain_actions: dict[str, float] = {}

    def _browser_endpoint(self, alias: str) -> str:
        return self.runtimes.endpoint(alias)

    def list_browsers(self) -> list[dict]:
        # Released generations remain in the runtime journal for cleanup/audit,
        # but are not live browser bindings and must not collide with a reused alias.
        return [
            browser
            for browser in self.runtimes.list_public()
            if browser.get("state") != "released"
        ]

    def info(self) -> dict:
        return {
            "service": "browser-gateway",
            "node_id": self.node_id,
        }

    def create(self, input: dict) -> dict:
        return self.runtimes.create(input)

    def change_state(self, alias: str, action: str, input: dict) -> None:
        generation = decode_generation(
            input, require_runtime=True, require_network=True
        )
        self.runtimes.change_state(alias, action, generation)

    def remove(self, alias: str, input: dict) -> None:
        generation = decode_generation(
            input,
            require_runtime=False,
            require_network=False,
            allow_profile_purge=True,
        )
        purge_profile = input.get("purge_profile", False)
        if type(purge_profile) is not bool:
            raise RequestError("purge_profile must be boolean", 400)
        self.runtimes.remove(alias, generation, purge_profile)

    def restore_proxy(self, alias: str, input: dict) -> None:
        validate_proxy_restore(input, alias)
        generation = decode_generation(
            {key: input.get(key) for key in ("binding_version", "runtime_id", "network_id")},
            True,
            True,
        )
        self.runtimes.restore_proxy(alias, generation, input["network_exit"])

    def _require_generation(self, alias: str, generation: dict) -> tuple[str, bool]:
        try:
            record = self.runtimes.require_generation(alias, generation)
        except FileNotFoundError:
            return "", False
        return record.runtime_id, True

    def _require_douyin_generation(self, alias: str, input: dict) -> None:
        record = self.runtimes.require_generation(alias, input)
        if input.get("network_exit_id", "") != record.network_exit_id:
            raise RequestError("runtime proxy generation does not match request", 409)

    def _alias_lock(self, alias: str):
        return self.runtimes.alias_lock(alias)

    def get_douyin(self, alias: str, input: dict) -> dict:
        if not valid_douyin_generation(input) or not valid_douyin_url(
            input.get("url", "")
        ):
            raise RequestError("invalid restricted browser request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                response = self.browser.get(alias, input["url"])
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin GET failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("restricted browser operation failed") from exc
        return {
            "status": response.status,
            "body": response.body,
            "challenge": response.challenge,
        }

    def resolve_douyin(self, alias: str, input: dict) -> dict:
        target = input.get("url", "")
        if not valid_douyin_generation(input) or not valid_douyin_source_url(target):
            raise RequestError("invalid restricted Douyin source URL", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                resolved = self.browser.resolve(alias, target)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin share resolution failed alias=%s reason=%s", alias, str(exc)
                )
                raise RequestError("restricted Douyin share resolution failed") from exc
        if not is_douyin_content_url(resolved):
            raise RequestError("Douyin share resolved to an unsupported URL", 502)
        return {"url": resolved}

    def get_douyin_media(self, alias: str, input: dict) -> dict:
        target = input.get("url", "")
        if not valid_douyin_generation(input) or not isinstance(target, str):
            raise RequestError("invalid restricted browser request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                response = self.browser.get_media(alias, target)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin media download failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("restricted browser media download failed") from exc
        return {
            "status": response.status,
            "content_type": response.content_type,
            "body_base64": response.body_base64,
        }

    def douyin_identity(self, alias: str, input: dict) -> dict:
        expected_account_key = input.get("expected_account_key", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_account_key, str)
            or not ACCOUNT_KEY_RE.fullmatch(expected_account_key)
        ):
            raise RequestError("invalid Douyin identity request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                identity = self.browser.identity(alias)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin identity verification failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError(
                    "Douyin login identity could not be verified"
                ) from exc
            if expected_account_key not in {
                identity["uid"],
                identity["sec_uid"],
                identity["unique_id"],
            }:
                raise RequestError(
                    "Douyin identity does not match the expected account", 409
                )
        return identity

    def douyin_login_qr(self, alias: str, input: dict) -> dict:
        if not valid_douyin_generation(input):
            raise RequestError("invalid Douyin login QR request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                screen = self.browser.login_qr(alias)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin login screen capture failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("Douyin login screen could not be captured") from exc
        return {
            "content_type": screen.content_type,
            "body_base64": screen.body_base64,
            "qr_detected": screen.qr_detected,
        }

    def douyin_message_history(self, alias: str, input: dict) -> dict:
        expected_uid = input.get("expected_uid", "")
        target_uid = input.get("target_uid", "")
        limit = input.get("limit", 100)
        cursor = input.get("cursor", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_uid, str)
            or not isinstance(target_uid, str)
            or not isinstance(cursor, str)
            or len(cursor) > 500
            or type(limit) is not int
            or not UID_RE.fullmatch(expected_uid)
            or not UID_RE.fullmatch(target_uid)
            or expected_uid == target_uid
            or not 1 <= limit <= 200
        ):
            raise RequestError("invalid Douyin message history request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                if cursor:
                    result = self.browser.message_history(
                        alias, expected_uid, target_uid, limit, cursor=cursor
                    )
                else:
                    result = self.browser.message_history(
                        alias, expected_uid, target_uid, limit
                    )
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin message history failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("Douyin message history failed") from exc
        return result

    def douyin_action(self, alias: str, input: dict) -> dict:
        expected_uid = input.get("expected_uid", "")
        action = input.get("action", "")
        target_uid = input.get("target_uid", "")
        target_comment_id = input.get("target_comment_id", "")
        target_work_id = input.get("target_work_id", "")
        text = input.get("text", "")
        confirm = input.get("confirm", False)
        operation_id = input.get("operation_id", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_uid, str)
            or not isinstance(action, str)
            or not isinstance(target_uid, str)
            or not isinstance(target_comment_id, str)
            or not isinstance(target_work_id, str)
            or not isinstance(text, str)
            or not isinstance(operation_id, str)
            or len(operation_id) > 200
            or type(confirm) is not bool
            or not UID_RE.fullmatch(expected_uid)
            or action not in ACTIONS
        ):
            raise RequestError("invalid Douyin action request", 400)
        if action == "repost":
            raise RequestError("ACTION_UNAVAILABLE", 409)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            self._claim_action(
                alias,
                str(input.get("runtime_id", "")),
                operation_id,
            )
            try:
                result = self.browser.action(
                    alias,
                    expected_uid,
                    action,
                    target_uid,
                    target_comment_id,
                    target_work_id,
                    text,
                    confirm,
                )
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                self._handle_douyin_action_error(alias, action, exc, operation_id)
                raise RequestError("Douyin action failed") from exc
            except Exception:
                self._release_action_ownership(alias, operation_id)
                raise
            else:
                self._release_action_ownership(alias, operation_id)
        return result

    def _handle_douyin_action_error(
        self, alias: str, action: str, error: DouyinError, operation_id: str = ""
    ) -> None:
        if getattr(error, "uncertain", False) or "timed out" in str(error).lower():
            self._retain_action_ownership(alias, operation_id)
        else:
            self._release_action_ownership(alias, operation_id)
        LOG.warning(
            "Douyin action failed alias=%s action=%s reason=%s",
            alias,
            action,
            str(error),
        )

    def _claim_action(
        self, alias: str, runtime_id: str = "", operation_id: str = ""
    ) -> None:
        now = time.monotonic()
        with self._action_ownership_lock:
            until = self._uncertain_actions.get(alias, 0.0)
            if until > now:
                raise RequestError("previous Douyin action outcome is uncertain", 409)
        if runtime_id:
            marker = self.browser.action_ownership(alias)
            if marker is not None:
                raise RequestError("previous Douyin action outcome is uncertain", 409)
            self.browser.set_action_ownership(
                alias,
                {
                    "runtime_id": runtime_id,
                    "operation_id": operation_id,
                    "started_at": time.time(),
                },
            )
        with self._action_ownership_lock:
            self._uncertain_actions.pop(alias, None)
            self._uncertain_actions[alias] = 0.0

    def _release_action_ownership(self, alias: str, operation_id: str = "") -> None:
        if operation_id:
            self.browser.clear_action_ownership(alias, operation_id)
        with self._action_ownership_lock:
            self._uncertain_actions.pop(alias, None)

    def _retain_action_ownership(self, alias: str, operation_id: str = "") -> None:
        with self._action_ownership_lock:
            # A timed-out page script may still finish its network write. Keep the
            # alias blocked until the browser generation is removed or replaced.
            self._uncertain_actions[alias] = math.inf

    def start_douyin_events(self, alias: str, input: dict) -> dict:
        expected_uid = input.get("expected_uid", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_uid, str)
            or not UID_RE.fullmatch(expected_uid)
        ):
            raise RequestError("invalid Douyin event request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                self.browser.identity(alias, expected_uid)
                return self.subscriptions.start(alias, expected_uid)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin event listener start failed alias=%s uid=%s reason=%s",
                    alias,
                    expected_uid,
                    exc,
                )
                raise RequestError("Douyin event listener could not start") from exc

    def poll_douyin_events(self, alias: str, input: dict, query: dict) -> list[dict]:
        if not valid_douyin_generation(input):
            raise RequestError("invalid Douyin event request", 400)
        try:
            limit = int(query.get("limit", ["50"])[0])
            wait = float(query.get("wait", ["25"])[0])
        except (IndexError, TypeError, ValueError) as exc:
            raise RequestError("invalid event poll options", 400) from exc
        if not 1 <= limit <= 100 or not 0 <= wait <= 30:
            raise RequestError("invalid event poll options", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
        try:
            acknowledgements = query.get("ack", [])
            delivery_ids = [
                value for raw in acknowledgements for value in raw.split(",")
            ]
            self.subscriptions.ack(alias, delivery_ids)
            return self.subscriptions.poll(alias, limit, wait)
        except DouyinError as exc:
            LOG.warning(
                "Douyin event listener poll failed alias=%s reason=%s",
                alias,
                exc,
            )
            raise RequestError("Douyin event listener is unavailable") from exc

    def stop_douyin_events(self, alias: str, input: dict) -> None:
        if not valid_douyin_generation(input):
            raise RequestError("invalid Douyin event request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            self.subscriptions.stop(alias)

class GatewayHTTPServer(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True

    gateway: Gateway

    def __init__(self, address, gateway: Gateway):
        super().__init__(address, GatewayHandler)
        self.gateway = gateway
        self._connections: set[socket.socket] = set()
        self._connections_lock = threading.Lock()
        self._connections_changed = threading.Condition(self._connections_lock)

    def process_request(self, request, client_address):
        with self._connections_changed:
            self._connections.add(cast(socket.socket, request))
        try:
            super().process_request(request, client_address)
        except Exception:
            with self._connections_changed:
                self._connections.discard(cast(socket.socket, request))
                self._connections_changed.notify_all()
            raise

    def process_request_thread(self, request, client_address):
        try:
            super().process_request_thread(request, client_address)
        finally:
            with self._connections_changed:
                self._connections.discard(cast(socket.socket, request))
                self._connections_changed.notify_all()

    def wait_for_requests(self, timeout: float) -> None:
        deadline = time.monotonic() + timeout
        with self._connections_changed:
            while self._connections:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
                self._connections_changed.wait(remaining)
            if self._connections:
                connections = list(self._connections)
            else:
                connections = []
        for connection in connections:
            with suppress(OSError):
                connection.shutdown(socket.SHUT_RDWR)
            connection.close()


class GatewayHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def setup(self) -> None:
        super().setup()
        self.request.settimeout(5.0)

    def do_GET(self) -> None:
        self._dispatch("GET")

    def do_POST(self) -> None:
        self._dispatch("POST")

    def do_DELETE(self) -> None:
        self._dispatch("DELETE")

    def log_message(self, format: str, *args) -> None:
        LOG.info("http_request %s", format % args)

    def _dispatch(self, method: str) -> None:
        parsed = urlsplit(self.path)
        if parsed.path == "/healthz":
            self._respond(204, b"")
            return
        if not parsed.path.startswith("/v1/"):
            self._respond(404, json_bytes({"error": "not found"}))
            return
        if not self._authorized():
            self._respond(401, json_bytes({"error": "gateway token rejected"}))
            return
        try:
            needs_body = method in {"POST", "DELETE"} or (
                method == "GET" and parsed.path.endswith("/douyin/events")
            )
            body = self._body() if needs_body else {}
            result = self._route(method, parsed.path, parse_qs(parsed.query), body)
            if result is None:
                self._respond(204, b"")
            elif isinstance(result, tuple):
                status, value = result
                if type(status) is not int:
                    raise RuntimeError("gateway route returned an invalid status")
                self._respond(status, json_bytes(value))
            else:
                self._respond(200, json_bytes(result))
        except (RuntimeError, OSError, ValueError, TypeError, KeyError) as exc:
            self._handle_exception(parsed.path, exc)

    def _handle_exception(self, path: str, exc: Exception) -> None:
        if isinstance(exc, RequestError):
            payload = {"error": str(exc)}
            if exc.network_id:
                payload["network_id"] = exc.network_id
            self._respond(exc.status, json_bytes(payload))
        elif isinstance(exc, FileNotFoundError):
            self._respond(404, json_bytes({"error": str(exc)}))
        elif isinstance(exc, BrowserRuntimeError):
            payload = {"error": str(exc)}
            if exc.network_id:
                payload["network_id"] = exc.network_id
            self._respond(exc.status, json_bytes(payload))
        elif isinstance(exc, ValueError):
            self._respond(400, json_bytes({"error": str(exc)}))
        else:
            LOG.exception("gateway request failed", extra={"path": path})
            self._respond(500, json_bytes({"error": "gateway operation failed"}))

    def _route(self, method: str, path: str, query: dict, body: dict):
        gateway = self.server_as_gateway().gateway
        if method == "GET" and path == "/v1/info":
            return gateway.info()
        if method == "GET" and path == "/v1/browsers":
            return gateway.list_browsers()
        if method == "POST" and path == "/v1/browsers":
            return 201, gateway.create(body)
        match = re.fullmatch(r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})", path)
        if match and method == "DELETE":
            gateway.remove(match.group(1), body)
            return None
        match = re.fullmatch(
            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/(start|stop|cancel|proxy)", path
        )
        if match:
            alias, action = match.groups()
            if method == "POST" and action in {"start", "stop", "cancel"}:
                gateway.change_state(alias, action, body)
                return None
            if method == "POST" and action == "proxy":
                gateway.restore_proxy(alias, body)
                return None
        match = re.fullmatch(
            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/douyin/(get|media|identity|resolve|login-qr|action|messages|events)",
            path,
        )
        if match:
            alias, action = match.groups()
            if action == "get" and method == "POST":
                return gateway.get_douyin(alias, body)
            if action == "resolve" and method == "POST":
                return gateway.resolve_douyin(alias, body)
            if action == "media" and method == "POST":
                return gateway.get_douyin_media(alias, body)
            if action == "identity" and method == "POST":
                return gateway.douyin_identity(alias, body)
            if action == "login-qr" and method == "POST":
                return gateway.douyin_login_qr(alias, body)
            if action == "action" and method == "POST":
                return gateway.douyin_action(alias, body)
            if action == "messages" and method == "POST":
                return gateway.douyin_message_history(alias, body)
            if action == "events":
                if method == "POST":
                    return gateway.start_douyin_events(alias, body)
                if method == "GET":
                    return gateway.poll_douyin_events(alias, body, query)
                if method == "DELETE":
                    gateway.stop_douyin_events(alias, body)
                    return None
        raise RequestError("not found", 404)

    def server_as_gateway(self) -> GatewayHTTPServer:
        if not isinstance(self.server, GatewayHTTPServer):
            raise TypeError("gateway HTTP server type is invalid")
        return self.server

    def _authorized(self) -> bool:
        supplied = self.headers.get("Authorization", "")
        return hmac.compare_digest(
            supplied, "Bearer " + self.server_as_gateway().gateway.token
        )

    def _body(self) -> dict:
        length_text = self.headers.get("Content-Length")
        if length_text is None:
            raise RequestError("request body is required", 400)
        try:
            length = int(length_text)
        except ValueError as exc:
            raise RequestError("invalid request body length", 400) from exc
        if length < 0 or length > 1 << 20:
            raise RequestError("request body is too large", 400)
        raw = self.rfile.read(length)
        try:
            value = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise RequestError("request body must be one JSON object", 400) from exc
        if not isinstance(value, dict):
            raise RequestError("request body must be one JSON object", 400)
        return value

    def _respond(self, status: int, body: bytes) -> None:
        try:
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            if body:
                self.wfile.write(body)
        except BrokenPipeError:
            LOG.info(
                "gateway client disconnected before response",
                extra={"status": status},
            )


def json_bytes(value: object) -> bytes:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()


def valid_douyin_source_url(raw: object) -> bool:
    return is_douyin_share_url(raw)


def valid_douyin_url(raw: object) -> bool:
    if not isinstance(raw, str):
        return False
    try:
        parsed = urlsplit(raw)
    except ValueError:
        return False
    if (
        parsed.scheme != "https"
        or parsed.netloc != "www.douyin.com"
        or parsed.username
        or parsed.fragment
    ):
        return False
    query = parse_qs(parsed.query, keep_blank_values=True)
    if parsed.path == DOUYIN_IDENTITY_PATH:
        return query == {"aid": ["6383"], "device_platform": ["webapp"]}
    if parsed.path == DOUYIN_PROFILE_OTHER_PATH:
        return valid_douyin_profile_query(query)
    if parsed.path == DOUYIN_WORKS_PATH:
        return valid_douyin_api_query(query, "sec_user_id", "max_cursor")
    if parsed.path == DOUYIN_WORK_DETAIL_PATH:
        return valid_douyin_work_detail_query(query)
    if parsed.path == DOUYIN_COMMENTS_PATH:
        return valid_douyin_api_query(query, "aweme_id", "cursor")
    return False


def valid_douyin_work_detail_query(query: dict[str, list[str]]) -> bool:
    return (
        len(query) == 3
        and query.get("aid") == ["6383"]
        and query.get("device_platform") == ["webapp"]
        and len(query.get("aweme_id", [])) == 1
        and bool(re.fullmatch(r"[1-9][0-9]{0,63}", query["aweme_id"][0]))
    )


def valid_douyin_profile_query(query: dict[str, list[str]]) -> bool:
    account_fields = [field for field in ("user_id", "sec_user_id") if field in query]
    return (
        len(query) == 3
        and query.get("aid") == ["6383"]
        and query.get("device_platform") == ["webapp"]
        and len(account_fields) == 1
        and valid_account_key_query(query, account_fields[0])
    )


def valid_douyin_api_query(
    query: dict[str, list[str]], account_field: str, cursor_field: str
) -> bool:
    return (
        len(query) == 5
        and query.get("aid") == ["6383"]
        and query.get("device_platform") == ["webapp"]
        and valid_account_key_query(query, account_field)
        and query.get("count") == ["20"]
        and numeric_cursor(query.get(cursor_field))
    )


def valid_account_key_query(query: dict[str, list[str]], key: str) -> bool:
    return len(query.get(key, [])) == 1 and bool(
        DOUYIN_ACCOUNT_KEY_RE.fullmatch(query[key][0])
    )


def numeric_cursor(values: list[str] | None) -> bool:
    if not values or len(values) != 1 or not values[0].isdigit():
        return False
    try:
        return int(values[0]) >= 0
    except ValueError:
        return False


def parse_proxy_exit(value: object) -> ProxyExit:
    try:
        return _parse_proxy_exit(value)
    except BrowserRuntimeError as exc:
        raise RequestError(str(exc), exc.status, exc.network_id) from exc


def validate_proxy_exit(exit: ProxyExit) -> None:
    try:
        _validate_proxy_exit(exit)
    except BrowserRuntimeError as exc:
        raise RequestError(str(exc), exc.status, exc.network_id) from exc


def validate_create(input: dict) -> None:
    try:
        validate_runtime_input(input)
    except BrowserRuntimeError as exc:
        raise RequestError(str(exc), exc.status, exc.network_id) from exc


def proxy_port(proxy_url: str) -> int:
    return urlsplit(proxy_url).port or 0


def decode_generation(
    value: dict,
    require_runtime: bool,
    require_network: bool,
    allow_profile_purge: bool = False,
) -> dict:
    allowed = {"binding_version", "runtime_id", "network_id"}
    if allow_profile_purge:
        allowed.add("purge_profile")
    if not isinstance(value, dict) or set(value) - allowed:
        raise RequestError(
            "binding_version, runtime_id and network_id must identify the expected generation",
            400,
        )
    binding = value.get("binding_version")
    runtime = value.get("runtime_id", "")
    network = value.get("network_id", "")
    if (
        type(binding) is not int
        or binding < 1
        or not isinstance(runtime, str)
        or not isinstance(network, str)
        or (
            runtime
            and runtime != RUNTIME_CLEANUP_SENTINEL
            and not RUNTIME_ID_RE.fullmatch(runtime)
        )
        or (network and not NETWORK_ID_RE.fullmatch(network))
        or (require_runtime and not runtime)
        or (require_network and not network)
    ):
        raise RequestError(
            "binding_version, runtime_id and network_id must identify the expected generation",
            400,
        )
    return {"binding_version": binding, "runtime_id": runtime, "network_id": network}


def validate_proxy_restore(value: dict, alias: str) -> None:
    del alias
    allowed = {
        "binding_version",
        "runtime_id",
        "network_id",
        "network_exit_id",
        "network_exit",
    }
    if not isinstance(value, dict) or set(value) - allowed:
        raise RequestError("invalid proxy recovery request", 400)
    generation = decode_generation(
        {
            key: value.get(key)
            for key in ("binding_version", "runtime_id", "network_id")
        },
        True,
        True,
    )
    exit_id = value.get("network_exit_id", "")
    if not isinstance(exit_id, str) or (exit_id and not EXIT_ID_RE.fullmatch(exit_id)):
        raise RequestError("invalid proxy recovery request", 400)
    exit = parse_proxy_exit(value.get("network_exit", {}))
    direct = not exit_id and exit == ProxyExit("", "", 0)
    if bool(exit_id) != (not direct):
        raise RequestError("invalid proxy recovery request", 400)
    if not direct:
        validate_proxy_exit(exit)
    value["network_exit"] = exit
    value.update(generation)
    value["network_exit_id"] = exit_id


def valid_douyin_generation(value: dict) -> bool:
    if not isinstance(value, dict):
        return False
    binding = value.get("binding_version")
    runtime = value.get("runtime_id", "")
    network = value.get("network_id", "")
    exit_id = value.get("network_exit_id", "")
    return (
        type(binding) is int
        and binding > 0
        and isinstance(runtime, str)
        and isinstance(network, str)
        and isinstance(exit_id, str)
        and bool(RUNTIME_ID_RE.fullmatch(runtime))
        and bool(NETWORK_ID_RE.fullmatch(network))
        and (not exit_id or bool(EXIT_ID_RE.fullmatch(exit_id)))
    )


def _positive_float(env: Mapping[str, str], key: str, default: float, maximum: float) -> float:
    raw = env.get(key, str(default)).strip()
    try:
        value = float(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{key} must be a number") from exc
    if not 0 < value <= maximum:
        raise ValueError(f"{key} is out of range")
    return value


def _optional_positive_int(env: Mapping[str, str], key: str) -> int | None:
    raw = env.get(key, "").strip()
    if not raw:
        return None
    try:
        value = int(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{key} must be an integer") from exc
    if value < 1:
        raise ValueError(f"{key} is out of range")
    return value


def _integer(env: Mapping[str, str], key: str, default: int) -> int:
    raw = env.get(key, str(default)).strip()
    try:
        return int(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{key} must be an integer") from exc


def load_config(env: Mapping[str, str] | None = None) -> dict:
    env = os.environ if env is None else env
    listen = env.get("LISTEN_ADDR", "0.0.0.0:8081").strip()
    token = env.get("GATEWAY_TOKEN", "").strip()
    state_dir = os.path.abspath(os.path.expanduser(env.get(
        "BROWSER_STATE_DIR", "~/.local/state/creatorhub/browser-gateway"
    ).strip()))
    profile_root = os.path.abspath(os.path.expanduser(env.get(
        "BROWSER_PROFILE_ROOT", "~/.local/share/creatorhub/browser-profiles"
    ).strip()))
    browser_path = os.path.abspath(os.path.expanduser(env.get(
        "BROWSER_PATH",
        "~/.local/share/creatorhub/browsers/fingerprint-chromium/148.0.7778.215/chrome",
    ).strip()))
    node_id = env.get("NODE_ID", "").strip()
    node_name = env.get("NODE_NAME", socket.gethostname()).strip()
    host, port = split_listen_address(listen)
    if len(token) < 16:
        raise ValueError("GATEWAY_TOKEN must be at least 16 characters")
    if not state_dir or not profile_root or not os.path.isabs(state_dir) or not os.path.isabs(profile_root):
        raise ValueError("BROWSER_STATE_DIR and BROWSER_PROFILE_ROOT must be absolute")
    if not node_id:
        node_id = ""
    elif not re.fullmatch(r"^[a-z0-9][a-z0-9._-]{0,63}$", node_id):
        raise ValueError("NODE_ID is invalid")
    if not node_name or has_control(node_name) or len(node_name) > 128:
        raise ValueError("NODE_NAME is invalid")
    if not 1 <= port <= 65535:
        raise ValueError("LISTEN_ADDR port must be 1..65535")
    external_display = _optional_positive_int(env, "RUNTIME_EXTERNAL_DISPLAY")
    try:
        available = os.path.isfile(browser_path) and os.access(browser_path, os.X_OK)
    except (OSError, TypeError) as exc:
        raise ValueError("browser executable is unavailable") from exc
    if not available:
        raise ValueError("browser executable is unavailable")
    return {
        "listen": (host, port),
        "state_dir": state_dir,
        "profile_root": profile_root,
        "browser_path": browser_path,
        "node_id": node_id,
        "node_name": node_name,
        "token": token,
        "cleanup_timeout": _positive_float(env, "RUNTIME_CLEANUP_TIMEOUT", 30.0, 300.0),
        "ready_timeout": _positive_float(env, "RUNTIME_READY_TIMEOUT", 60.0, 300.0),
        "min_free_bytes": _integer(env, "RUNTIME_MIN_FREE_BYTES", 20 * 1024**3),
        "log_max_bytes": _integer(env, "RUNTIME_LOG_MAX_BYTES", 1 * 1024**3),
        "profile_cache_max_bytes": _integer(env, "PROFILE_CACHE_MAX_BYTES", 20 * 1024**3),
        "external_display": external_display,
    }


def split_listen_address(value: str) -> tuple[str, int]:
    if value.startswith(":"):
        host, port_text = "", value[1:]
    elif value.startswith("["):
        closing = value.find("]:" )
        if closing <= 1:
            raise ValueError("LISTEN_ADDR must be host:port")
        host, port_text = value[1:closing], value[closing + 2:]
    else:
        if ":" not in value:
            raise ValueError("LISTEN_ADDR must be host:port")
        host, port_text = value.rsplit(":", 1)
    try:
        port = int(port_text)
    except ValueError as exc:
        raise ValueError("LISTEN_ADDR port must be an integer") from exc
    return host, port


def load_stable_node_id(state_dir: str, configured: str) -> str:
    if configured:
        return configured
    path = Path(state_dir) / "node-id"
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    if path.exists():
        value = path.read_text(encoding="utf-8").strip()
        if not re.fullmatch(r"^[a-z0-9][a-z0-9._-]{0,63}$", value):
            raise ValueError("persisted NODE_ID is invalid")
        return value
    value = "node-" + secrets.token_hex(16)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as output:
            output.write(value + "\n")
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary, path)
    finally:
        with suppress(FileNotFoundError):
            temporary.unlink()
    return value


def run() -> None:
    config = load_config()
    if os.geteuid() == 0:
        raise RuntimeError("browser gateway must run as a non-root user")
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    node_id = load_stable_node_id(config["state_dir"], config["node_id"])
    runtimes = NativeRuntimeManager(
        state_dir=config["state_dir"],
        profile_root=config["profile_root"],
        node_id=node_id,
        browser_path=config["browser_path"],
        cleanup_timeout=config["cleanup_timeout"],
        ready_timeout=config["ready_timeout"],
        min_free_bytes=config["min_free_bytes"],
        log_max_bytes=config["log_max_bytes"],
        profile_cache_max_bytes=config["profile_cache_max_bytes"],
        external_display=config["external_display"],
    )
    gateway = Gateway(runtimes, config["token"], node_id)
    server = GatewayHTTPServer(config["listen"], gateway)
    LOG.info(
        json.dumps(
            {
                "service": "browser-gateway",
                "node_id": node_id,
                "node_name": config["node_name"],
                "listen_addr": f"{config['listen'][0]}:{config['listen'][1]}",
                "browser_path": config["browser_path"],
                "external_display": config["external_display"],
            },
            ensure_ascii=False,
        )
    )
    shutdown_requested = threading.Event()

    def request_shutdown(_signum, _frame) -> None:
        if shutdown_requested.is_set():
            return
        shutdown_requested.set()
        threading.Thread(
            target=server.shutdown,
            name="gateway-shutdown",
            daemon=True,
        ).start()

    signal.signal(signal.SIGINT, request_shutdown)
    signal.signal(signal.SIGTERM, request_shutdown)
    try:
        server.serve_forever()
    finally:
        server.wait_for_requests(30.0)
        gateway.subscriptions.close()
        runtimes.close()
        server.server_close()


if __name__ == "__main__":
    run()