from __future__ import annotations

import asyncio
import hashlib
import json
import math
import re
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import uuid4

from pymongo import ASCENDING, DESCENDING, GEOSPHERE, ReturnDocument
from pymongo.errors import DuplicateKeyError, OperationFailure

from wbb import BOT_PROFILE_ID, control_db

profilesdb = control_db.directory_profiles
locationsdb = control_db.directory_locations
membershipsdb = control_db.directory_memberships
eventsdb = control_db.directory_events
settingsdb = control_db.directory_settings
geocode_cachedb = control_db.directory_geocode_cache
geocode_jobsdb = control_db.directory_geocode_jobs

APPLICATION_NONE = "none"
APPLICATION_PENDING = "pending"
APPLICATION_APPROVED = "approved"
APPLICATION_REJECTED = "rejected"
APPLICATION_REVOKED = "revoked"
APPLICATION_STATUSES = {
    APPLICATION_NONE,
    APPLICATION_PENDING,
    APPLICATION_APPROVED,
    APPLICATION_REJECTED,
    APPLICATION_REVOKED,
}

PRESENCE_ONLINE = "online"
PRESENCE_OFFLINE = "offline"

DEFAULT_DIRECTORY_SETTINGS: dict[str, Any] = {
    "nominatim_enabled": True,
    "nominatim_endpoint": "https://nominatim.openstreetmap.org/reverse",
    "nominatim_contact": "",
    "nominatim_user_agent": "TelegramBotAdmin/1.0",
    "online_duration_hours": 24,
    "nearby_radius_km": 20,
    "nearby_radius_options_km": [5, 10, 20, 50],
}

_index_lock = asyncio.Lock()
_indexes_ready = False


class DirectoryDataError(ValueError):
    def __init__(self, code: str, message: str):
        super().__init__(message)
        self.code = code


def utc_now() -> datetime:
    return datetime.now(UTC)


def aware_utc(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=UTC)
    return value.astimezone(UTC)


def display_name(
    first_name: str | None,
    last_name: str | None,
    username: str | None = None,
    user_id: int | None = None,
) -> str:
    name = " ".join(
        value.strip()
        for value in (first_name, last_name)
        if value and value.strip()
    )
    if name:
        return name
    if username:
        return f"@{username}"
    return f"用户 {user_id}" if user_id is not None else "未知用户"


async def ensure_directory_indexes() -> None:
    global _indexes_ready
    if _indexes_ready:
        return
    async with _index_lock:
        if _indexes_ready:
            return
        await profilesdb.create_index([("user_id", ASCENDING)], unique=True)
        await profilesdb.create_index(
            [
                ("application_status", ASCENDING),
                ("listed", ASCENDING),
                ("updated_at", DESCENDING),
            ]
        )
        await profilesdb.create_index(
            [("presence_status", ASCENDING), ("online_until", ASCENDING)]
        )
        await locationsdb.create_index([("user_id", ASCENDING)], unique=True)
        await locationsdb.create_index([("point", GEOSPHERE)])
        await locationsdb.create_index([("updated_at", DESCENDING)])
        await membershipsdb.create_index(
            [
                ("bot_id", ASCENDING),
                ("chat_id", ASCENDING),
                ("user_id", ASCENDING),
            ],
            unique=True,
        )
        await membershipsdb.create_index(
            [("user_id", ASCENDING), ("active", ASCENDING), ("verified_at", DESCENDING)]
        )
        await eventsdb.create_index([("event_id", ASCENDING)], unique=True)
        await eventsdb.create_index(
            [("idempotency_key", ASCENDING)],
            unique=True,
            sparse=True,
        )
        await eventsdb.create_index([("created_at", DESCENDING)])
        await eventsdb.create_index([("user_id", ASCENDING), ("created_at", DESCENDING)])
        await settingsdb.create_index([("settings_id", ASCENDING)], unique=True)
        await geocode_cachedb.create_index([("coordinate_key", ASCENDING)], unique=True)
        await geocode_jobsdb.create_index([("user_id", ASCENDING)], unique=True)
        await geocode_jobsdb.create_index(
            [("status", ASCENDING), ("next_attempt_at", ASCENDING)]
        )
        _indexes_ready = True


async def get_directory_settings() -> dict[str, Any]:
    await ensure_directory_indexes()
    stored = await settingsdb.find_one({"settings_id": "global"}) or {}
    return {
        **DEFAULT_DIRECTORY_SETTINGS,
        **{
            key: value
            for key, value in stored.items()
            if key not in {"_id", "settings_id"}
        },
    }


async def set_directory_settings(values: dict[str, Any]) -> dict[str, Any]:
    await ensure_directory_indexes()
    current = await get_directory_settings()
    target = {**current, **values}
    endpoint = str(target.get("nominatim_endpoint") or "").strip()
    if not endpoint.startswith(("https://", "http://127.0.0.1", "http://localhost")):
        raise DirectoryDataError(
            "invalid_nominatim_endpoint",
            "Nominatim 地址必须使用 HTTPS；仅本机自建服务可以使用 HTTP。",
        )
    try:
        radius_options = sorted(
            {
                int(value)
                for value in target.get("nearby_radius_options_km", [])
                if int(value) > 0
            }
        )
        default_radius = int(target.get("nearby_radius_km") or 0)
    except (TypeError, ValueError) as exc:
        raise DirectoryDataError(
            "invalid_radius_options", "附近范围必须是正整数。"
        ) from exc
    if not radius_options:
        raise DirectoryDataError("invalid_radius_options", "附近范围选项不能为空。")
    if default_radius not in radius_options:
        raise DirectoryDataError(
            "invalid_default_radius", "默认附近范围必须属于可选范围。"
        )
    contact = str(target.get("nominatim_contact") or "").strip()
    user_agent = str(
        target.get("nominatim_user_agent") or "TelegramBotAdmin/1.0"
    ).strip()
    if "\n" in contact or "\r" in contact or "\n" in user_agent or "\r" in user_agent:
        raise DirectoryDataError(
            "invalid_nominatim_identity",
            "Nominatim 联系信息和 User-Agent 不能包含换行符。",
        )
    if contact and "@" not in contact and not contact.startswith(("https://", "http://")):
        raise DirectoryDataError(
            "invalid_nominatim_contact",
            "Nominatim 联系信息必须是邮箱或 URL。",
        )
    target.update(
        {
            "nominatim_enabled": bool(target.get("nominatim_enabled")),
            "nominatim_endpoint": endpoint,
            "nominatim_contact": contact,
            "nominatim_user_agent": user_agent,
            "online_duration_hours": 24,
            "nearby_radius_km": default_radius,
            "nearby_radius_options_km": radius_options,
            "updated_at": utc_now(),
        }
    )
    await settingsdb.update_one(
        {"settings_id": "global"},
        {"$set": target, "$setOnInsert": {"created_at": utc_now()}},
        upsert=True,
    )
    return await get_directory_settings()


async def record_directory_event(
    *,
    event_type: str,
    user_id: int,
    actor_id: int | str | None,
    actor_name: str = "",
    source: str,
    reason: str = "",
    metadata: dict[str, Any] | None = None,
    idempotency_key: str | None = None,
    bot_id: str | None = None,
    chat_id: int | None = None,
) -> dict[str, Any] | None:
    await ensure_directory_indexes()
    document = {
        "event_id": uuid4().hex,
        "event_type": event_type,
        "user_id": int(user_id),
        "actor_id": actor_id,
        "actor_name": actor_name,
        "source": source,
        "reason": reason,
        "metadata": metadata or {},
        "bot_id": bot_id or str(BOT_PROFILE_ID),
        "chat_id": int(chat_id) if chat_id is not None else None,
        "created_at": utc_now(),
    }
    if idempotency_key:
        document["idempotency_key"] = idempotency_key
    try:
        await eventsdb.insert_one(document)
    except DuplicateKeyError:
        return None
    return document


async def upsert_directory_identity(
    *,
    user_id: int,
    username: str | None,
    first_name: str | None,
    last_name: str | None,
    source: str = "telegram",
) -> dict[str, Any]:
    await ensure_directory_indexes()
    normalized_username = str(username or "").lstrip("@").strip() or None
    now = utc_now()
    previous = await profilesdb.find_one({"user_id": int(user_id)})
    identity = {
        "username": normalized_username,
        "first_name": str(first_name or "").strip() or None,
        "last_name": str(last_name or "").strip() or None,
        "display_name": display_name(
            first_name,
            last_name,
            normalized_username,
            int(user_id),
        ),
    }
    updates: dict[str, Any] = {**identity, "updated_at": now}
    username_removed = bool(previous and previous.get("username") and not normalized_username)
    if username_removed and previous.get("application_status") == APPLICATION_APPROVED:
        updates.update(
            {
                "listed": False,
                "presence_status": PRESENCE_OFFLINE,
                "online_until": None,
            }
        )
    await profilesdb.update_one(
        {"user_id": int(user_id)},
        {
            "$set": updates,
            "$setOnInsert": {
                "application_status": APPLICATION_NONE,
                "listed": False,
                "presence_status": PRESENCE_OFFLINE,
                "created_at": now,
            },
        },
        upsert=True,
    )
    if username_removed:
        await record_directory_event(
            event_type="teacher_username_removed",
            user_id=int(user_id),
            actor_id=int(user_id),
            actor_name=identity["display_name"],
            source=source,
            reason="老师移除了 Telegram 用户名，系统自动下榜并下线。",
        )
    return await get_directory_profile(int(user_id)) or {}


async def get_directory_profile(user_id: int) -> dict[str, Any] | None:
    await ensure_directory_indexes()
    return await profilesdb.find_one({"user_id": int(user_id)})


async def get_directory_location(user_id: int) -> dict[str, Any] | None:
    await ensure_directory_indexes()
    return await locationsdb.find_one({"user_id": int(user_id)})


def validate_coordinates(longitude: float, latitude: float) -> tuple[float, float]:
    try:
        lon = float(longitude)
        lat = float(latitude)
    except (TypeError, ValueError) as exc:
        raise DirectoryDataError("invalid_coordinates", "经纬度必须是数字。") from exc
    if not math.isfinite(lon) or not -180 <= lon <= 180:
        raise DirectoryDataError("invalid_longitude", "经度必须在 -180 到 180 之间。")
    if not math.isfinite(lat) or not -90 <= lat <= 90:
        raise DirectoryDataError("invalid_latitude", "纬度必须在 -90 到 90 之间。")
    return lon, lat


def coordinate_key(longitude: float, latitude: float) -> str:
    raw = json.dumps(
        [float(longitude), float(latitude)],
        ensure_ascii=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(raw.encode("ascii")).hexdigest()


async def save_directory_location(
    *,
    user_id: int,
    longitude: float,
    latitude: float,
    source: str,
    actor_id: int | str,
    actor_name: str,
    geocode_consent: bool | None = None,
    bot_id: str | None = None,
) -> dict[str, Any]:
    await ensure_directory_indexes()
    lon, lat = validate_coordinates(longitude, latitude)
    previous = await get_directory_location(int(user_id))
    consent = (
        geocode_consent
        if geocode_consent is not None
        else previous.get("geocode_consent") if previous else None
    )
    key = coordinate_key(lon, lat)
    now = utc_now()
    status = "pending" if consent is True else "disabled" if consent is False else "consent_required"
    await locationsdb.update_one(
        {"user_id": int(user_id)},
        {
            "$set": {
                "point": {"type": "Point", "coordinates": [lon, lat]},
                "longitude": lon,
                "latitude": lat,
                "coordinate_key": key,
                "geocode_consent": consent,
                "geocode_status": status,
                "region": None,
                "source": source,
                "source_bot_id": bot_id or str(BOT_PROFILE_ID),
                "updated_at": now,
            },
            "$setOnInsert": {"created_at": now},
        },
        upsert=True,
    )
    await profilesdb.update_one(
        {"user_id": int(user_id)},
        {
            "$set": {"location_updated_at": now, "updated_at": now},
            "$setOnInsert": {
                "application_status": APPLICATION_NONE,
                "listed": False,
                "presence_status": PRESENCE_OFFLINE,
                "created_at": now,
            },
        },
        upsert=True,
    )
    await record_directory_event(
        event_type="location_updated",
        user_id=int(user_id),
        actor_id=actor_id,
        actor_name=actor_name,
        source=source,
        reason="更新位置",
        metadata={"geocode_consent": consent},
        bot_id=bot_id,
    )
    if consent is True:
        await enqueue_geocode_job(int(user_id))
    return await get_directory_location(int(user_id)) or {}


async def set_geocode_consent(
    *,
    user_id: int,
    consent: bool,
    source: str,
    actor_id: int | str,
    actor_name: str,
) -> dict[str, Any]:
    location = await get_directory_location(int(user_id))
    if not location:
        raise DirectoryDataError("location_required", "请先更新位置。")
    updates: dict[str, Any] = {
        "geocode_consent": bool(consent),
        "geocode_status": "pending" if consent else "disabled",
        "updated_at": utc_now(),
    }
    if not consent:
        updates["region"] = None
        await geocode_jobsdb.delete_one({"user_id": int(user_id)})
    await locationsdb.update_one({"user_id": int(user_id)}, {"$set": updates})
    await record_directory_event(
        event_type="geocode_consent_updated",
        user_id=int(user_id),
        actor_id=actor_id,
        actor_name=actor_name,
        source=source,
        reason="同意区域解析" if consent else "关闭区域解析",
        metadata={"geocode_consent": bool(consent)},
    )
    if consent:
        await enqueue_geocode_job(int(user_id))
    return await get_directory_location(int(user_id)) or {}


async def clear_directory_location(
    *,
    user_id: int,
    actor_id: int | str,
    actor_name: str,
    source: str,
    reason: str,
) -> bool:
    await ensure_directory_indexes()
    result = await locationsdb.delete_one({"user_id": int(user_id)})
    await geocode_jobsdb.delete_one({"user_id": int(user_id)})
    await profilesdb.update_one(
        {"user_id": int(user_id)},
        {
            "$set": {
                "listed": False,
                "presence_status": PRESENCE_OFFLINE,
                "online_until": None,
                "location_updated_at": None,
                "updated_at": utc_now(),
            }
        },
    )
    if result.deleted_count:
        await record_directory_event(
            event_type="location_cleared",
            user_id=int(user_id),
            actor_id=actor_id,
            actor_name=actor_name,
            source=source,
            reason=reason,
        )
    return bool(result.deleted_count)


async def submit_teacher_application(
    *,
    user_id: int,
    source: str,
    bot_id: str,
) -> dict[str, Any]:
    await ensure_directory_indexes()
    profile = await get_directory_profile(int(user_id))
    location = await get_directory_location(int(user_id))
    if not profile or not profile.get("username"):
        raise DirectoryDataError(
            "username_required", "申请老师前必须先设置 Telegram 用户名。"
        )
    if not location:
        raise DirectoryDataError("location_required", "申请老师前必须先更新位置。")
    if profile.get("application_status") == APPLICATION_APPROVED:
        raise DirectoryDataError("already_teacher", "你已经是老师，无需重复申请。")
    if profile.get("application_status") == APPLICATION_REVOKED:
        raise DirectoryDataError(
            "teacher_revoked", "老师资格已被撤销，只能由管理员重新批准。"
        )
    if profile.get("application_status") == APPLICATION_PENDING:
        return profile
    now = utc_now()
    await profilesdb.update_one(
        {"user_id": int(user_id)},
        {
            "$set": {
                "application_status": APPLICATION_PENDING,
                "application_reason": "",
                "application_source_bot_id": str(bot_id),
                "applied_at": now,
                "decided_at": None,
                "decided_by": None,
                "updated_at": now,
            }
        },
    )
    await record_directory_event(
        event_type="teacher_application_submitted",
        user_id=int(user_id),
        actor_id=int(user_id),
        actor_name=str(profile.get("display_name") or user_id),
        source=source,
        reason="申请成为老师",
        bot_id=bot_id,
    )
    return await get_directory_profile(int(user_id)) or {}


async def decide_teacher_application(
    *,
    user_id: int,
    action: str,
    actor_id: int | str,
    actor_name: str,
    source: str,
    reason: str = "",
    authorization_chat_id: int | None = None,
) -> dict[str, Any]:
    await ensure_directory_indexes()
    profile = await get_directory_profile(int(user_id))
    if not profile:
        raise DirectoryDataError("profile_not_found", "未找到该成员资料。")
    action_status = {
        "approve": APPLICATION_APPROVED,
        "reject": APPLICATION_REJECTED,
        "revoke": APPLICATION_REVOKED,
    }
    if action not in action_status:
        raise DirectoryDataError("invalid_decision", "不支持的审批操作。")
    if action in {"reject", "revoke"} and not reason.strip():
        raise DirectoryDataError("reason_required", "拒绝或撤销时必须填写原因。")
    if action == "approve":
        if not profile.get("username"):
            raise DirectoryDataError(
                "username_required", "该成员没有 Telegram 用户名，无法批准。"
            )
        if not await get_directory_location(int(user_id)):
            raise DirectoryDataError("location_required", "该成员尚未更新位置。")
    now = utc_now()
    updates: dict[str, Any] = {
        "application_status": action_status[action],
        "application_reason": reason.strip(),
        "decided_at": now,
        "decided_by": actor_id,
        "decided_by_name": actor_name,
        "authorization_chat_id": authorization_chat_id,
        "updated_at": now,
    }
    if action != "approve":
        updates.update(
            {
                "listed": False,
                "presence_status": PRESENCE_OFFLINE,
                "online_until": None,
            }
        )
    await profilesdb.update_one({"user_id": int(user_id)}, {"$set": updates})
    await record_directory_event(
        event_type=f"teacher_application_{action}",
        user_id=int(user_id),
        actor_id=actor_id,
        actor_name=actor_name,
        source=source,
        reason=reason,
        metadata={"authorization_chat_id": authorization_chat_id},
        chat_id=authorization_chat_id,
    )
    return await get_directory_profile(int(user_id)) or {}


async def set_teacher_state(
    *,
    user_id: int,
    action: str,
    actor_id: int | str,
    actor_name: str,
    source: str,
    reason: str = "",
    idempotency_key: str | None = None,
    chat_id: int | None = None,
) -> tuple[dict[str, Any], bool]:
    await ensure_directory_indexes()
    profile = await get_directory_profile(int(user_id))
    if not profile or profile.get("application_status") != APPLICATION_APPROVED:
        raise DirectoryDataError("teacher_required", "该成员尚未取得老师资格。")
    location = await get_directory_location(int(user_id))
    if action in {"list", "online"}:
        if not profile.get("username"):
            raise DirectoryDataError("username_required", "请先设置 Telegram 用户名。")
        if not location:
            raise DirectoryDataError("location_required", "请先更新位置。")
    if action == "online" and not profile.get("listed"):
        raise DirectoryDataError("listing_required", "请先上榜，再切换为上线状态。")
    if action not in {"list", "unlist", "online", "offline"}:
        raise DirectoryDataError("invalid_teacher_action", "不支持的老师状态操作。")
    event = await record_directory_event(
        event_type=f"teacher_{action}",
        user_id=int(user_id),
        actor_id=actor_id,
        actor_name=actor_name,
        source=source,
        reason=reason or f"老师{action}",
        idempotency_key=idempotency_key,
        chat_id=chat_id,
    )
    applied = event is not None
    now = utc_now()
    updates: dict[str, Any] = {"updated_at": now}
    if action == "list":
        updates["listed"] = True
        updates["listed_at"] = now
    elif action == "unlist":
        updates.update(
            {
                "listed": False,
                "presence_status": PRESENCE_OFFLINE,
                "online_until": None,
            }
        )
    elif action == "online":
        updates.update(
            {
                "presence_status": PRESENCE_ONLINE,
                "online_until": now + timedelta(hours=24),
                "last_online_at": now,
            }
        )
    else:
        updates.update(
            {"presence_status": PRESENCE_OFFLINE, "online_until": None}
        )
    await profilesdb.update_one({"user_id": int(user_id)}, {"$set": updates})
    return await get_directory_profile(int(user_id)) or {}, applied


async def expire_teacher_presence() -> int:
    await ensure_directory_indexes()
    now = utc_now()
    expired = 0
    while True:
        item = await profilesdb.find_one_and_update(
            {
                "presence_status": PRESENCE_ONLINE,
                "online_until": {"$lte": now},
            },
            {
                "$set": {
                    "presence_status": PRESENCE_OFFLINE,
                    "online_until": None,
                    "updated_at": now,
                }
            },
            sort=[("online_until", ASCENDING)],
            return_document=ReturnDocument.BEFORE,
        )
        if not item:
            break
        expired += 1
        previous_until = aware_utc(item.get("online_until"))
        await record_directory_event(
            event_type="teacher_auto_offline",
            user_id=int(item["user_id"]),
            actor_id="system",
            actor_name="系统",
            source="scheduler",
            reason="上线状态已超过 24 小时。",
            idempotency_key=(
                f"teacher-auto-offline:{item['user_id']}:"
                f"{previous_until.isoformat() if previous_until else 'unknown'}"
            ),
        )
    return expired


def teacher_is_online(profile: dict[str, Any], *, now: datetime | None = None) -> bool:
    current = now or utc_now()
    until = aware_utc(profile.get("online_until"))
    return bool(
        profile.get("presence_status") == PRESENCE_ONLINE
        and until
        and until > current
    )


def _haversine_meters(
    longitude_a: float,
    latitude_a: float,
    longitude_b: float,
    latitude_b: float,
) -> float:
    radius = 6_371_008.8
    lon_a, lat_a, lon_b, lat_b = map(
        math.radians,
        (longitude_a, latitude_a, longitude_b, latitude_b),
    )
    delta_lon = lon_b - lon_a
    delta_lat = lat_b - lat_a
    value = (
        math.sin(delta_lat / 2) ** 2
        + math.cos(lat_a) * math.cos(lat_b) * math.sin(delta_lon / 2) ** 2
    )
    return 2 * radius * math.asin(math.sqrt(value))


async def _list_teachers_fallback(
    *,
    longitude: float,
    latitude: float,
    max_distance_meters: float | None,
    query: str,
    page: int,
    page_size: int,
) -> tuple[list[dict[str, Any]], int]:
    profile_filter: dict[str, Any] = {
        "application_status": APPLICATION_APPROVED,
        "listed": True,
        "username": {"$nin": [None, ""]},
    }
    if query:
        pattern = re.compile(re.escape(query.lstrip("@")), re.IGNORECASE)
        profile_filter["$or"] = [
            {"username": pattern},
            {"display_name": pattern},
            {"user_id": int(query)} if query.isdigit() else {"user_id": -1},
        ]
    profiles = {
        int(item["user_id"]): item
        async for item in profilesdb.find(profile_filter)
    }
    values: list[dict[str, Any]] = []
    if profiles:
        async for location in locationsdb.find(
            {"user_id": {"$in": list(profiles)}}
        ):
            point = location.get("point", {}).get("coordinates") or []
            if len(point) != 2:
                continue
            distance = _haversine_meters(
                longitude,
                latitude,
                float(point[0]),
                float(point[1]),
            )
            if max_distance_meters is not None and distance > max_distance_meters:
                continue
            values.append(
                {
                    **profiles[int(location["user_id"])],
                    "distance_meters": distance,
                    "region": location.get("region"),
                    "location_updated_at": location.get("updated_at"),
                }
            )
    values.sort(key=lambda item: (item["distance_meters"], int(item["user_id"])))
    total = len(values)
    start = (page - 1) * page_size
    return values[start : start + page_size], total


async def list_directory_teachers(
    *,
    longitude: float,
    latitude: float,
    max_distance_meters: float | None = None,
    query: str = "",
    page: int = 1,
    page_size: int = 10,
) -> tuple[list[dict[str, Any]], int]:
    await ensure_directory_indexes()
    lon, lat = validate_coordinates(longitude, latitude)
    profile_match: dict[str, Any] = {
        "profile.application_status": APPLICATION_APPROVED,
        "profile.listed": True,
        "profile.username": {"$nin": [None, ""]},
    }
    if query:
        pattern = re.compile(re.escape(query.lstrip("@")), re.IGNORECASE)
        profile_match["$or"] = [
            {"profile.username": pattern},
            {"profile.display_name": pattern},
            {"profile.user_id": int(query)} if query.isdigit() else {"profile.user_id": -1},
        ]
    geo_near: dict[str, Any] = {
        "near": {"type": "Point", "coordinates": [lon, lat]},
        "distanceField": "distance_meters",
        "spherical": True,
        "key": "point",
    }
    if max_distance_meters is not None:
        geo_near["maxDistance"] = float(max_distance_meters)
    pipeline = [
        {"$geoNear": geo_near},
        {
            "$lookup": {
                "from": profilesdb.name,
                "localField": "user_id",
                "foreignField": "user_id",
                "as": "profile",
            }
        },
        {"$unwind": "$profile"},
        {"$match": profile_match},
        {"$sort": {"distance_meters": 1, "user_id": 1}},
        {
            "$facet": {
                "items": [
                    {"$skip": (max(1, page) - 1) * max(1, page_size)},
                    {"$limit": max(1, page_size)},
                ],
                "count": [{"$count": "total"}],
            }
        },
    ]
    try:
        results = await locationsdb.aggregate(pipeline).to_list(length=1)
        result = results[0] if results else {"items": [], "count": []}
        items = [
            {
                **item["profile"],
                "distance_meters": item["distance_meters"],
                "region": item.get("region"),
                "location_updated_at": item.get("updated_at"),
            }
            for item in result.get("items", [])
        ]
        count = result.get("count") or []
        return items, int(count[0]["total"]) if count else 0
    except (NotImplementedError, OperationFailure):
        return await _list_teachers_fallback(
            longitude=lon,
            latitude=lat,
            max_distance_meters=max_distance_meters,
            query=query,
            page=page,
            page_size=page_size,
        )


async def list_teacher_applications(
    *,
    status: str = "",
    query: str = "",
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[dict[str, Any]], int]:
    await ensure_directory_indexes()
    filters: dict[str, Any] = {}
    if status:
        if status not in APPLICATION_STATUSES:
            raise DirectoryDataError("invalid_application_status", "申请状态无效。")
        filters["application_status"] = status
    else:
        filters["application_status"] = {"$ne": APPLICATION_NONE}
    if query:
        pattern = re.compile(re.escape(query.lstrip("@")), re.IGNORECASE)
        filters["$or"] = [
            {"username": pattern},
            {"display_name": pattern},
            {"user_id": int(query)} if query.isdigit() else {"user_id": -1},
        ]
    total = await profilesdb.count_documents(filters)
    items = await (
        profilesdb.find(filters)
        .sort([("applied_at", DESCENDING), ("updated_at", DESCENDING)])
        .skip((page - 1) * page_size)
        .limit(page_size)
        .to_list(length=page_size)
    )
    location_map = {
        int(item["user_id"]): item
        async for item in locationsdb.find(
            {"user_id": {"$in": [int(item["user_id"]) for item in items]}}
        )
    }
    for item in items:
        location = location_map.get(int(item["user_id"]))
        item["location"] = (
            {
                "region": location.get("region"),
                "geocode_status": location.get("geocode_status"),
                "updated_at": location.get("updated_at"),
            }
            if location
            else None
        )
        item["online"] = teacher_is_online(item)
    return items, total


async def list_directory_profiles(
    *,
    query: str = "",
    application_status: str = "",
    listed: bool | None = None,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[dict[str, Any]], int]:
    filters: dict[str, Any] = {}
    if application_status:
        filters["application_status"] = application_status
    if listed is not None:
        filters["listed"] = listed
    if query:
        pattern = re.compile(re.escape(query.lstrip("@")), re.IGNORECASE)
        filters["$or"] = [
            {"username": pattern},
            {"display_name": pattern},
            {"user_id": int(query)} if query.isdigit() else {"user_id": -1},
        ]
    total = await profilesdb.count_documents(filters)
    items = await (
        profilesdb.find(filters)
        .sort([("updated_at", DESCENDING)])
        .skip((page - 1) * page_size)
        .limit(page_size)
        .to_list(length=page_size)
    )
    locations = {
        int(item["user_id"]): item
        async for item in locationsdb.find(
            {"user_id": {"$in": [int(item["user_id"]) for item in items]}}
        )
    }
    for item in items:
        location = locations.get(int(item["user_id"]))
        item["location"] = (
            {
                "region": location.get("region"),
                "geocode_status": location.get("geocode_status"),
                "updated_at": location.get("updated_at"),
            }
            if location
            else None
        )
        item["online"] = teacher_is_online(item)
    return items, total


async def list_directory_locations(
    *,
    query: str = "",
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[dict[str, Any]], int]:
    await ensure_directory_indexes()
    filters: dict[str, Any] = {}
    if query:
        pattern = re.compile(re.escape(query.lstrip("@")), re.IGNORECASE)
        profile_filters = {
            "$or": [
                {"username": pattern},
                {"display_name": pattern},
                {"user_id": int(query)} if query.isdigit() else {"user_id": -1},
            ]
        }
        user_ids = [
            int(item["user_id"])
            async for item in profilesdb.find(profile_filters, {"user_id": 1})
        ]
        filters["user_id"] = {"$in": user_ids}
    total = await locationsdb.count_documents(filters)
    items = await (
        locationsdb.find(filters)
        .sort("updated_at", DESCENDING)
        .skip((page - 1) * page_size)
        .limit(page_size)
        .to_list(length=page_size)
    )
    profiles = {
        int(item["user_id"]): item
        async for item in profilesdb.find(
            {"user_id": {"$in": [int(item["user_id"]) for item in items]}}
        )
    }
    for item in items:
        item["profile"] = profiles.get(int(item["user_id"]), {})
    return items, total


async def list_directory_events(
    *,
    query: str = "",
    event_type: str = "",
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[dict[str, Any]], int]:
    filters: dict[str, Any] = {}
    if event_type:
        filters["event_type"] = event_type
    if query:
        pattern = re.compile(re.escape(query), re.IGNORECASE)
        filters["$or"] = [
            {"actor_name": pattern},
            {"reason": pattern},
            {"user_id": int(query)} if query.isdigit() else {"user_id": -1},
        ]
    total = await eventsdb.count_documents(filters)
    items = await (
        eventsdb.find(filters)
        .sort("created_at", DESCENDING)
        .skip((page - 1) * page_size)
        .limit(page_size)
        .to_list(length=page_size)
    )
    return items, total


async def upsert_directory_membership(
    *,
    bot_id: str,
    chat_id: int,
    user_id: int,
    status: str,
    active: bool,
    chat_title: str = "",
    username: str | None = None,
    display_name_value: str = "",
    verified: bool = False,
) -> dict[str, Any]:
    await ensure_directory_indexes()
    now = utc_now()
    values: dict[str, Any] = {
        "status": status,
        "active": bool(active),
        "username": username,
        "display_name": display_name_value,
        "observed_at": now,
        "updated_at": now,
    }
    if chat_title:
        values["chat_title"] = chat_title
    if verified:
        values["verified_at"] = now
    await membershipsdb.update_one(
        {
            "bot_id": str(bot_id),
            "chat_id": int(chat_id),
            "user_id": int(user_id),
        },
        {"$set": values, "$setOnInsert": {"created_at": now}},
        upsert=True,
    )
    return await membershipsdb.find_one(
        {
            "bot_id": str(bot_id),
            "chat_id": int(chat_id),
            "user_id": int(user_id),
        }
    ) or {}


async def list_membership_candidates(user_id: int) -> list[dict[str, Any]]:
    await ensure_directory_indexes()
    return await membershipsdb.find(
        {"user_id": int(user_id), "active": True}
    ).sort("verified_at", DESCENDING).to_list(length=200)


async def enqueue_geocode_job(user_id: int) -> dict[str, Any] | None:
    await ensure_directory_indexes()
    location = await get_directory_location(int(user_id))
    if not location or location.get("geocode_consent") is not True:
        return None
    key = str(location["coordinate_key"])
    cached = await geocode_cachedb.find_one({"coordinate_key": key})
    if cached:
        await locationsdb.update_one(
            {"user_id": int(user_id), "coordinate_key": key},
            {
                "$set": {
                    "region": cached.get("region"),
                    "geocode_status": "completed",
                    "geocoded_at": cached.get("created_at"),
                    "updated_at": utc_now(),
                }
            },
        )
        return None
    now = utc_now()
    await geocode_jobsdb.update_one(
        {"user_id": int(user_id)},
        {
            "$set": {
                "coordinate_key": key,
                "longitude": float(location["longitude"]),
                "latitude": float(location["latitude"]),
                "status": "pending",
                "attempts": 0,
                "next_attempt_at": now,
                "updated_at": now,
            },
            "$setOnInsert": {"created_at": now},
        },
        upsert=True,
    )
    return await geocode_jobsdb.find_one({"user_id": int(user_id)})


async def claim_geocode_job() -> dict[str, Any] | None:
    await ensure_directory_indexes()
    now = utc_now()
    return await geocode_jobsdb.find_one_and_update(
        {
            "status": "pending",
            "$or": [
                {"next_attempt_at": {"$lte": now}},
                {"next_attempt_at": {"$exists": False}},
            ],
        },
        {
            "$set": {
                "status": "processing",
                "claimed_at": now,
                "updated_at": now,
            },
            "$inc": {"attempts": 1},
        },
        sort=[("created_at", ASCENDING)],
        return_document=ReturnDocument.AFTER,
    )


async def complete_geocode_job(
    *,
    job: dict[str, Any],
    region: dict[str, Any],
) -> None:
    now = utc_now()
    key = str(job["coordinate_key"])
    await geocode_cachedb.update_one(
        {"coordinate_key": key},
        {
            "$setOnInsert": {
                "coordinate_key": key,
                "region": region,
                "provider": "nominatim",
                "created_at": now,
            }
        },
        upsert=True,
    )
    await locationsdb.update_one(
        {
            "user_id": int(job["user_id"]),
            "coordinate_key": key,
            "geocode_consent": True,
        },
        {
            "$set": {
                "region": region,
                "geocode_status": "completed",
                "geocoded_at": now,
                "updated_at": now,
            }
        },
    )
    await geocode_jobsdb.delete_one(
        {"user_id": int(job["user_id"]), "coordinate_key": key}
    )


async def fail_geocode_job(
    *,
    job: dict[str, Any],
    error: str,
    retry_after_seconds: int = 60,
) -> None:
    attempts = int(job.get("attempts") or 1)
    key_filter = {
        "user_id": int(job["user_id"]),
        "coordinate_key": str(job["coordinate_key"]),
    }
    if attempts >= 3:
        await geocode_jobsdb.update_one(
            key_filter,
            {
                "$set": {
                    "status": "failed",
                    "error": error[:500],
                    "updated_at": utc_now(),
                }
            },
        )
        await locationsdb.update_one(
            key_filter,
            {
                "$set": {
                    "geocode_status": "failed",
                    "geocode_error": error[:500],
                    "updated_at": utc_now(),
                }
            },
        )
        return
    await geocode_jobsdb.update_one(
        key_filter,
        {
            "$set": {
                "status": "pending",
                "error": error[:500],
                "next_attempt_at": utc_now()
                + timedelta(seconds=max(1, retry_after_seconds)),
                "updated_at": utc_now(),
            }
        },
    )