feat(监控): 完善系统监控、登录状态与访问审计能力

backend/app/services/geo_location_metadata.py

@@ -0,0 +1,175 @@
+"""Canonical map metadata for monitoring source locations.
+
+ip2region provides administrative names but no coordinates. This module keeps
+the normalization and centroid contract on the server so web and desktop
+clients render the same location semantics.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.services.ip_location import IpLocation
+
+
+@dataclass(frozen=True)
+class GeoLocationMetadata:
+    country: str
+    country_code: str
+    region_code: str
+    longitude: float | None
+    latitude: float | None
+    accuracy_level: str
+
+
+COUNTRY_ALIASES = {
+    "中国": "China",
+    "China": "China",
+    "Mainland China": "China",
+    "中国香港": "China",
+    "中国澳门": "China",
+    "中国台湾": "China",
+    "美国": "United States",
+    "United States": "United States",
+    "United States of America": "United States",
+    "USA": "United States",
+    "土耳其": "Türkiye",
+    "Turkey": "Türkiye",
+    "Türkiye": "Türkiye",
+}
+
+COUNTRY_CODES = {
+    "China": "CN",
+    "United States": "US",
+    "Netherlands": "NL",
+    "Türkiye": "TR",
+    "Australia": "AU",
+    "Japan": "JP",
+    "Singapore": "SG",
+    "Germany": "DE",
+    "France": "FR",
+    "United Kingdom": "GB",
+    "Canada": "CA",
+    "India": "IN",
+    "Russia": "RU",
+}
+
+COUNTRY_CENTROIDS = {
+    "China": (104.1954, 35.8617),
+    "United States": (-95.7129, 37.0902),
+    "Netherlands": (5.2913, 52.1326),
+    "Türkiye": (35.2433, 38.9637),
+    "Australia": (133.7751, -25.2744),
+    "Japan": (138.2529, 36.2048),
+    "Singapore": (103.8198, 1.3521),
+    "Germany": (10.4515, 51.1657),
+    "France": (2.2137, 46.2276),
+    "United Kingdom": (-3.436, 55.3781),
+    "Canada": (-106.3468, 56.1304),
+    "India": (78.9629, 20.5937),
+    "Russia": (105.3188, 61.524),
+}
+
+CHINA_REGION_METADATA = {
+    "北京市": ("CN-BJ", 116.4074, 39.9042),
+    "天津市": ("CN-TJ", 117.2008, 39.0842),
+    "河北省": ("CN-HE", 114.5025, 38.0455),
+    "山西省": ("CN-SX", 112.5492, 37.857),
+    "内蒙古自治区": ("CN-NM", 111.6708, 40.8183),
+    "辽宁省": ("CN-LN", 123.4315, 41.8057),
+    "吉林省": ("CN-JL", 125.3245, 43.8868),
+    "黑龙江省": ("CN-HL", 126.6424, 45.7567),
+    "上海市": ("CN-SH", 121.4737, 31.2304),
+    "江苏省": ("CN-JS", 118.7633, 32.0617),
+    "浙江省": ("CN-ZJ", 120.1551, 30.2741),
+    "安徽省": ("CN-AH", 117.2272, 31.8206),
+    "福建省": ("CN-FJ", 119.2965, 26.0745),
+    "江西省": ("CN-JX", 115.8582, 28.682),
+    "山东省": ("CN-SD", 117.1201, 36.6512),
+    "河南省": ("CN-HA", 113.6254, 34.7466),
+    "湖北省": ("CN-HB", 114.3055, 30.5928),
+    "湖南省": ("CN-HN", 112.9388, 28.2282),
+    "广东省": ("CN-GD", 113.2644, 23.1291),
+    "广西壮族自治区": ("CN-GX", 108.3669, 22.817),
+    "海南省": ("CN-HI", 110.3312, 20.0311),
+    "重庆": ("CN-CQ", 106.5516, 29.563),
+    "重庆市": ("CN-CQ", 106.5516, 29.563),
+    "四川省": ("CN-SC", 104.0665, 30.5723),
+    "贵州省": ("CN-GZ", 106.6302, 26.647),
+    "云南省": ("CN-YN", 102.8329, 24.8801),
+    "西藏自治区": ("CN-XZ", 91.1322, 29.6604),
+    "陕西省": ("CN-SN", 108.9398, 34.3416),
+    "甘肃省": ("CN-GS", 103.8343, 36.0611),
+    "青海省": ("CN-QH", 101.7782, 36.6171),
+    "宁夏回族自治区": ("CN-NX", 106.2309, 38.4872),
+    "新疆维吾尔自治区": ("CN-XJ", 87.6168, 43.8256),
+    "台湾省": ("CN-TW", 121.5654, 25.033),
+    "香港特别行政区": ("CN-HK", 114.1694, 22.3193),
+    "澳门特别行政区": ("CN-MO", 113.5439, 22.1987),
+}
+
+CITY_CENTROIDS = {
+    "南京": (118.7969, 32.0603),
+    "南京市": (118.7969, 32.0603),
+    "San Jose": (-121.8863, 37.3382),
+    "South Holland": (4.493, 52.0208),
+    "Istanbul": (28.9784, 41.0082),
+}
+
+
+def resolve_geo_location_metadata(location: IpLocation) -> GeoLocationMetadata:
+    if location.location in {"局域网", "本机"}:
+        return GeoLocationMetadata(
+            country="",
+            country_code="PRIVATE",
+            region_code="PRIVATE",
+            longitude=None,
+            latitude=None,
+            accuracy_level="private",
+        )
+
+    country = COUNTRY_ALIASES.get(location.country, location.country)
+    country_code = COUNTRY_CODES.get(country, "")
+
+    city_coordinate = CITY_CENTROIDS.get(location.city)
+    if city_coordinate:
+        return GeoLocationMetadata(
+            country=country,
+            country_code=country_code,
+            region_code="",
+            longitude=city_coordinate[0],
+            latitude=city_coordinate[1],
+            accuracy_level="city",
+        )
+
+    region_metadata = CHINA_REGION_METADATA.get(location.province)
+    if country in {"China", "中国"} and region_metadata:
+        region_code, longitude, latitude = region_metadata
+        return GeoLocationMetadata(
+            country="China",
+            country_code="CN",
+            region_code=region_code,
+            longitude=longitude,
+            latitude=latitude,
+            accuracy_level="region",
+        )
+
+    country_coordinate = COUNTRY_CENTROIDS.get(country)
+    if country_coordinate:
+        return GeoLocationMetadata(
+            country=country,
+            country_code=country_code,
+            region_code="",
+            longitude=country_coordinate[0],
+            latitude=country_coordinate[1],
+            accuracy_level="country",
+        )
+
+    return GeoLocationMetadata(
+        country=country,
+        country_code=country_code,
+        region_code="",
+        longitude=None,
+        latitude=None,
+        accuracy_level="unknown",
+    )

backend/app/services/ip_location.py

@@ -9,6 +9,7 @@ import ipaddress
 import logging
 import sys
 from dataclasses import dataclass
+from functools import lru_cache
 from pathlib import Path
 from typing import Optional
 
@@ -125,5 +126,6 @@ class Ip2RegionResolver:
 _resolver = Ip2RegionResolver()
 
 
+@lru_cache(maxsize=8192)
 def resolve_ip_location(ip: str | None) -> IpLocation:
     return _resolver.lookup(ip)

backend/app/services/monitoring_retention.py

@@ -0,0 +1,151 @@
+"""监测数据留存清理任务。"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from sqlalchemy import delete
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.permission_access_log import PermissionAccessLog
+from app.models.permission_metric_snapshot import PermissionMetricSnapshot
+from app.models.security_access_log import SecurityAccessLog
+from app.models.source_location_snapshot import SourceLocationSnapshot
+from app.models.user_login_session import UserLoginSession
+
+logger = logging.getLogger("ctms.monitoring_retention")
+
+
+@dataclass
+class MonitoringRetentionState:
+    running: bool = False
+    last_run_started_at: datetime | None = None
+    last_success_at: datetime | None = None
+    last_error_at: datetime | None = None
+    last_error_type: str | None = None
+    error_count: int = 0
+    last_deleted: dict[str, int] = field(default_factory=dict)
+    total_deleted: dict[str, int] = field(default_factory=dict)
+
+    def snapshot(self) -> dict[str, Any]:
+        return {
+            "running": self.running,
+            "access_log_retention_days": settings.MONITORING_ACCESS_LOG_RETENTION_DAYS,
+            "metric_retention_days": settings.MONITORING_METRIC_RETENTION_DAYS,
+            "login_activity_retention_days": settings.USER_LOGIN_ACTIVITY_RETENTION_DAYS,
+            "interval_seconds": settings.MONITORING_RETENTION_INTERVAL_SECONDS,
+            "last_run_started_at": (
+                self.last_run_started_at.isoformat() if self.last_run_started_at else None
+            ),
+            "last_success_at": self.last_success_at.isoformat() if self.last_success_at else None,
+            "last_error_at": self.last_error_at.isoformat() if self.last_error_at else None,
+            "last_error_type": self.last_error_type,
+            "error_count": self.error_count,
+            "last_deleted": dict(self.last_deleted),
+            "total_deleted": dict(self.total_deleted),
+        }
+
+
+_state = MonitoringRetentionState()
+
+
+def get_monitoring_retention_status() -> dict[str, Any]:
+    return _state.snapshot()
+
+
+def _deleted_count(result: Any) -> int:
+    rowcount = getattr(result, "rowcount", 0)
+    return max(0, int(rowcount or 0))
+
+
+async def purge_expired_monitoring_data(
+    *, now: datetime | None = None
+) -> dict[str, int]:
+    reference_time = now or datetime.now(timezone.utc)
+    access_cutoff = reference_time - timedelta(
+        days=settings.MONITORING_ACCESS_LOG_RETENTION_DAYS
+    )
+    metric_cutoff = reference_time - timedelta(
+        days=settings.MONITORING_METRIC_RETENTION_DAYS
+    )
+    login_activity_cutoff = reference_time - timedelta(
+        days=settings.USER_LOGIN_ACTIVITY_RETENTION_DAYS
+    )
+
+    async with SessionLocal() as session:
+        permission_result = await session.execute(
+            delete(PermissionAccessLog).where(PermissionAccessLog.created_at < access_cutoff)
+        )
+        security_result = await session.execute(
+            delete(SecurityAccessLog).where(SecurityAccessLog.created_at < access_cutoff)
+        )
+        metric_result = await session.execute(
+            delete(PermissionMetricSnapshot).where(
+                PermissionMetricSnapshot.bucket_time < metric_cutoff
+            )
+        )
+        source_location_result = await session.execute(
+            delete(SourceLocationSnapshot).where(
+                SourceLocationSnapshot.bucket_time < access_cutoff
+            )
+        )
+        login_session_result = await session.execute(
+            delete(UserLoginSession).where(UserLoginSession.last_seen_at < login_activity_cutoff)
+        )
+        await session.commit()
+
+    return {
+        "permission_access_logs": _deleted_count(permission_result),
+        "security_access_logs": _deleted_count(security_result),
+        "permission_metric_snapshots": _deleted_count(metric_result),
+        "source_location_snapshots": _deleted_count(source_location_result),
+        "user_login_sessions": _deleted_count(login_session_result),
+    }
+
+
+async def run_monitoring_retention_once(
+    *, now: datetime | None = None
+) -> dict[str, int]:
+    _state.last_run_started_at = now or datetime.now(timezone.utc)
+    try:
+        deleted = await purge_expired_monitoring_data(now=now)
+    except Exception as exc:
+        _state.last_error_at = datetime.now(timezone.utc)
+        _state.last_error_type = type(exc).__name__
+        _state.error_count += 1
+        raise
+
+    _state.last_success_at = datetime.now(timezone.utc)
+    _state.last_deleted = deleted
+    for key, count in deleted.items():
+        _state.total_deleted[key] = _state.total_deleted.get(key, 0) + count
+    return deleted
+
+
+async def run_monitoring_retention(stop_event: asyncio.Event) -> None:
+    logger.info("Monitoring retention task started")
+    _state.running = True
+    try:
+        while not stop_event.is_set():
+            try:
+                deleted = await run_monitoring_retention_once()
+                if any(deleted.values()):
+                    logger.info("Purged expired monitoring data: %s", deleted)
+            except Exception:
+                logger.exception("Failed to purge expired monitoring data")
+
+            try:
+                await asyncio.wait_for(
+                    stop_event.wait(),
+                    timeout=settings.MONITORING_RETENTION_INTERVAL_SECONDS,
+                )
+            except asyncio.TimeoutError:
+                continue
+    finally:
+        _state.running = False
+        logger.info("Monitoring retention task stopped")

backend/app/services/monitoring_server_location.py

@@ -0,0 +1,164 @@
+"""Resolve the monitoring deployment's public network location.
+
+The map server marker is derived from the deployment's public IP instead of a
+frontend or configuration coordinate. An explicit public IP is accepted for
+restricted networks; otherwise the public frontend hostname and, finally, a
+small set of public-IP discovery endpoints are used. Results are cached so the
+monitoring API never performs per-request network discovery.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import logging
+import socket
+import time
+from dataclasses import dataclass
+from urllib.parse import urlparse
+
+import httpx
+
+from app.core.config import settings
+from app.services.geo_location_metadata import resolve_geo_location_metadata
+from app.services.ip_location import resolve_ip_location
+
+logger = logging.getLogger("ctms.monitoring_server_location")
+
+
+@dataclass(frozen=True)
+class MonitoringServerLocation:
+    name: str
+    longitude: float
+    latitude: float
+    accuracy_level: str
+    resolution_source: str
+
+    def to_public_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "longitude": self.longitude,
+            "latitude": self.latitude,
+            "accuracy_level": self.accuracy_level,
+            "resolution_source": self.resolution_source,
+        }
+
+
+_cached_location: MonitoringServerLocation | None = None
+_cache_initialized = False
+_cache_expires_at = 0.0
+_cache_lock = asyncio.Lock()
+
+
+def _normalize_public_ip(value: str | None) -> str | None:
+    candidate = (value or "").strip().splitlines()[0] if (value or "").strip() else ""
+    try:
+        address = ipaddress.ip_address(candidate)
+    except ValueError:
+        return None
+    return str(address) if address.is_global else None
+
+
+async def _resolve_frontend_public_ip() -> str | None:
+    hostname = urlparse(settings.FRONTEND_PUBLIC_URL).hostname
+    if not hostname:
+        return None
+    literal_ip = _normalize_public_ip(hostname)
+    if literal_ip:
+        return literal_ip
+    try:
+        loop = asyncio.get_running_loop()
+        records = await loop.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
+    except (OSError, socket.gaierror):
+        return None
+    candidates = []
+    for _, _, _, _, socket_address in records:
+        candidate = _normalize_public_ip(str(socket_address[0]))
+        if candidate and candidate not in candidates:
+            candidates.append(candidate)
+    return next((item for item in candidates if ":" not in item), candidates[0] if candidates else None)
+
+
+async def _discover_public_ip() -> tuple[str | None, str]:
+    configured_ip = _normalize_public_ip(settings.MONITORING_SERVER_PUBLIC_IP)
+    if configured_ip:
+        return configured_ip, "configured_public_ip"
+
+    frontend_ip = await _resolve_frontend_public_ip()
+    if frontend_ip:
+        return frontend_ip, "frontend_dns"
+
+    if settings.ENV == "test":
+        return None, "unavailable"
+
+    urls = [
+        item.strip()
+        for item in settings.MONITORING_PUBLIC_IP_DISCOVERY_URLS.split(",")
+        if item.strip()
+    ]
+    timeout = httpx.Timeout(settings.MONITORING_PUBLIC_IP_DISCOVERY_TIMEOUT_SECONDS)
+    async with httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client:
+        for url in urls:
+            try:
+                response = await client.get(url, headers={"Accept": "text/plain"})
+                response.raise_for_status()
+            except httpx.HTTPError:
+                logger.warning("Public IP discovery endpoint unavailable", extra={"endpoint": url})
+                continue
+            discovered_ip = _normalize_public_ip(response.text[:128])
+            if discovered_ip:
+                return discovered_ip, "public_ip_discovery"
+    return None, "unavailable"
+
+
+async def resolve_monitoring_server_location(*, force: bool = False) -> MonitoringServerLocation | None:
+    global _cached_location, _cache_initialized, _cache_expires_at
+
+    now = time.monotonic()
+    if not force and _cache_initialized and now < _cache_expires_at:
+        return _cached_location
+
+    async with _cache_lock:
+        now = time.monotonic()
+        if not force and _cache_initialized and now < _cache_expires_at:
+            return _cached_location
+
+        public_ip, resolution_source = await _discover_public_ip()
+        location = None
+        if public_ip:
+            ip_location = resolve_ip_location(public_ip)
+            metadata = resolve_geo_location_metadata(ip_location)
+            if metadata.longitude is not None and metadata.latitude is not None:
+                country = metadata.country or ip_location.country
+                name = " / ".join(
+                    part for part in [country, ip_location.province, ip_location.city] if part
+                ) or "公网服务器"
+                location = MonitoringServerLocation(
+                    name=name,
+                    longitude=metadata.longitude,
+                    latitude=metadata.latitude,
+                    accuracy_level=metadata.accuracy_level,
+                    resolution_source=resolution_source,
+                )
+                logger.info(
+                    "Monitoring server location resolved",
+                    extra={"resolution_source": resolution_source, "accuracy_level": metadata.accuracy_level},
+                )
+            else:
+                logger.warning("Deployment public IP has no usable geo coordinates")
+        else:
+            logger.warning("Unable to discover the deployment public IP")
+
+        _cached_location = location
+        _cache_initialized = True
+        cache_seconds = settings.MONITORING_SERVER_LOCATION_CACHE_SECONDS if location else 300
+        _cache_expires_at = now + cache_seconds
+        return location
+
+
+def reset_monitoring_server_location_cache() -> None:
+    """Reset process-local state for configuration reloads and tests."""
+    global _cached_location, _cache_initialized, _cache_expires_at
+    _cached_location = None
+    _cache_initialized = False
+    _cache_expires_at = 0.0

backend/app/services/permission_log_writer.py

@@ -13,6 +13,7 @@ import asyncio
 import logging
 import uuid
 from datetime import datetime, timezone
+from typing import Any
 
 from app.db.session import SessionLocal
 from app.models.permission_access_log import PermissionAccessLog
@@ -22,85 +23,150 @@ logger = logging.getLogger("ctms.permission_log_writer")
 BATCH_SIZE = 50
 FLUSH_INTERVAL = 5.0
 QUEUE_MAX_SIZE = 10000
+WRITE_ATTEMPTS = 2
+_QUEUE_STOP = object()
 
 
 class PermissionLogWriter:
 
     def __init__(self):
-        self._queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
+        self._queue: asyncio.Queue[dict | object] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
         self._task: asyncio.Task | None = None
+        self._stopping = False
+        self._started_at: datetime | None = None
+        self._last_success_at: datetime | None = None
+        self._last_error_at: datetime | None = None
+        self._last_error_type: str | None = None
+        self._accepted_entries = 0
+        self._written_entries = 0
+        self._dropped_entries = 0
+        self._failed_batches = 0
+        self._failed_entries = 0
+        self._retry_count = 0
 
     def enqueue(self, entry: dict) -> None:
+        if self._stopping:
+            self._dropped_entries += 1
+            logger.warning("Permission log writer is stopping, dropping entry")
+            return
         try:
             self._queue.put_nowait(entry)
+            self._accepted_entries += 1
         except asyncio.QueueFull:
+            self._dropped_entries += 1
             logger.warning("Permission log queue full, dropping entry")
 
     async def start(self) -> None:
+        if self._task and not self._task.done():
+            return
+        self._stopping = False
+        self._started_at = datetime.now(timezone.utc)
         self._task = asyncio.create_task(self._flush_loop())
         logger.info("PermissionLogWriter started")
 
     async def stop(self) -> None:
-        if self._task:
-            self._task.cancel()
-            try:
-                await self._task
-            except asyncio.CancelledError:
-                pass
-        await self._drain()
+        if self._task and not self._task.done():
+            self._stopping = True
+            await self._queue.put(_QUEUE_STOP)
+            await self._task
+        self._task = None
         logger.info("PermissionLogWriter stopped")
 
     async def _flush_loop(self) -> None:
         while True:
-            batch = await self._collect_batch()
+            batch, should_stop = await self._collect_batch()
             if batch:
                 await self._write_batch(batch)
+            if should_stop:
+                break
 
-    async def _collect_batch(self) -> list[dict]:
+    async def _collect_batch(self) -> tuple[list[dict], bool]:
         batch: list[dict] = []
         try:
             first = await asyncio.wait_for(self._queue.get(), timeout=FLUSH_INTERVAL)
+            if first is _QUEUE_STOP:
+                return batch, True
             batch.append(first)
         except asyncio.TimeoutError:
-            return batch
+            return batch, False
 
         while len(batch) < BATCH_SIZE:
             try:
                 item = self._queue.get_nowait()
+                if item is _QUEUE_STOP:
+                    return batch, True
                 batch.append(item)
             except asyncio.QueueEmpty:
                 break
-        return batch
+        return batch, False
 
-    async def _write_batch(self, batch: list[dict]) -> None:
-        try:
-            async with SessionLocal() as session:
-                for entry in batch:
-                    log = PermissionAccessLog(
-                        id=uuid.uuid4(),
-                        study_id=entry["study_id"],
-                        user_id=entry["user_id"],
-                        endpoint_key=entry["endpoint_key"],
-                        role=entry["role"],
-                        allowed=entry["allowed"],
-                        elapsed_ms=entry["elapsed_ms"],
-                        ip_address=entry.get("ip_address"),
-                        created_at=entry.get("created_at", datetime.now(timezone.utc)),
-                    )
-                    session.add(log)
-                await session.commit()
-        except Exception:
-            logger.exception("Failed to write permission access log batch (%d entries)", len(batch))
-
-    async def _drain(self) -> None:
-        batch: list[dict] = []
-        while not self._queue.empty():
+    async def _write_batch(self, batch: list[dict]) -> bool:
+        for attempt in range(WRITE_ATTEMPTS):
             try:
-                batch.append(self._queue.get_nowait())
-            except asyncio.QueueEmpty:
-                break
-        if batch:
-            await self._write_batch(batch)
+                async with SessionLocal() as session:
+                    for entry in batch:
+                        log = PermissionAccessLog(
+                            id=uuid.uuid4(),
+                            study_id=entry["study_id"],
+                            user_id=entry["user_id"],
+                            endpoint_key=entry["endpoint_key"],
+                            role=entry["role"],
+                            allowed=entry["allowed"],
+                            elapsed_ms=entry["elapsed_ms"],
+                            ip_address=entry.get("ip_address"),
+                            user_agent=entry.get("user_agent"),
+                            client_type=entry.get("client_type"),
+                            client_version=entry.get("client_version"),
+                            client_platform=entry.get("client_platform"),
+                            build_channel=entry.get("build_channel"),
+                            build_commit=entry.get("build_commit"),
+                            request_headers=entry.get("request_headers"),
+                            request_snapshot=entry.get("request_snapshot"),
+                            request_id=entry.get("request_id"),
+                            created_at=entry.get("created_at", datetime.now(timezone.utc)),
+                        )
+                        session.add(log)
+                    await session.commit()
+            except Exception as exc:
+                self._last_error_at = datetime.now(timezone.utc)
+                self._last_error_type = type(exc).__name__
+                if attempt + 1 < WRITE_ATTEMPTS:
+                    self._retry_count += 1
+                    logger.warning(
+                        "Permission access log batch write failed; retrying (%d entries)",
+                        len(batch),
+                    )
+                    await asyncio.sleep(0)
+                    continue
+                self._failed_batches += 1
+                self._failed_entries += len(batch)
+                logger.exception(
+                    "Failed to write permission access log batch after retries (%d entries)",
+                    len(batch),
+                )
+                return False
+            self._written_entries += len(batch)
+            self._last_success_at = datetime.now(timezone.utc)
+            return True
+        return False
+
+    def stats(self) -> dict[str, Any]:
+        return {
+            "running": bool(self._task and not self._task.done()),
+            "stopping": self._stopping,
+            "queue_size": self._queue.qsize(),
+            "queue_capacity": self._queue.maxsize,
+            "accepted_entries": self._accepted_entries,
+            "written_entries": self._written_entries,
+            "dropped_entries": self._dropped_entries,
+            "failed_batches": self._failed_batches,
+            "failed_entries": self._failed_entries,
+            "retry_count": self._retry_count,
+            "started_at": self._started_at.isoformat() if self._started_at else None,
+            "last_success_at": self._last_success_at.isoformat() if self._last_success_at else None,
+            "last_error_at": self._last_error_at.isoformat() if self._last_error_at else None,
+            "last_error_type": self._last_error_type,
+        }
 
 
 _writer: PermissionLogWriter | None = None

backend/app/services/security_access_log_writer.py

@@ -6,99 +6,170 @@ import asyncio
 import logging
 import uuid
 from datetime import datetime, timezone
+from typing import Any
 
 from app.db.session import SessionLocal
 from app.models.security_access_log import SecurityAccessLog
+from app.services.ip_location import resolve_ip_location
+from app.services.security_events import classify_security_event
 
 logger = logging.getLogger("ctms.security_access_log_writer")
 
 BATCH_SIZE = 100
 FLUSH_INTERVAL = 3.0
 QUEUE_MAX_SIZE = 20000
+WRITE_ATTEMPTS = 2
+_QUEUE_STOP = object()
 
 
 class SecurityAccessLogWriter:
     def __init__(self) -> None:
-        self._queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
+        self._queue: asyncio.Queue[dict | object] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
         self._task: asyncio.Task | None = None
+        self._stopping = False
+        self._started_at: datetime | None = None
+        self._last_success_at: datetime | None = None
+        self._last_error_at: datetime | None = None
+        self._last_error_type: str | None = None
+        self._accepted_entries = 0
+        self._written_entries = 0
+        self._dropped_entries = 0
+        self._failed_batches = 0
+        self._failed_entries = 0
+        self._retry_count = 0
 
     def enqueue(self, entry: dict) -> None:
+        if self._stopping:
+            self._dropped_entries += 1
+            logger.warning("Security access log writer is stopping, dropping entry")
+            return
         try:
             self._queue.put_nowait(entry)
+            self._accepted_entries += 1
         except asyncio.QueueFull:
+            self._dropped_entries += 1
             logger.warning("Security access log queue full, dropping entry")
 
     async def start(self) -> None:
+        if self._task and not self._task.done():
+            return
+        self._stopping = False
+        self._started_at = datetime.now(timezone.utc)
         self._task = asyncio.create_task(self._flush_loop())
         logger.info("SecurityAccessLogWriter started")
 
     async def stop(self) -> None:
-        if self._task:
-            self._task.cancel()
-            try:
-                await self._task
-            except asyncio.CancelledError:
-                pass
-        await self._drain()
+        if self._task and not self._task.done():
+            self._stopping = True
+            await self._queue.put(_QUEUE_STOP)
+            await self._task
+        self._task = None
         logger.info("SecurityAccessLogWriter stopped")
 
     async def _flush_loop(self) -> None:
         while True:
-            batch = await self._collect_batch()
+            batch, should_stop = await self._collect_batch()
             if batch:
                 await self._write_batch(batch)
+            if should_stop:
+                break
 
-    async def _collect_batch(self) -> list[dict]:
+    async def _collect_batch(self) -> tuple[list[dict], bool]:
         batch: list[dict] = []
         try:
             first = await asyncio.wait_for(self._queue.get(), timeout=FLUSH_INTERVAL)
+            if first is _QUEUE_STOP:
+                return batch, True
             batch.append(first)
         except asyncio.TimeoutError:
-            return batch
+            return batch, False
 
         while len(batch) < BATCH_SIZE:
             try:
-                batch.append(self._queue.get_nowait())
+                item = self._queue.get_nowait()
+                if item is _QUEUE_STOP:
+                    return batch, True
+                batch.append(item)
             except asyncio.QueueEmpty:
                 break
-        return batch
+        return batch, False
 
-    async def _write_batch(self, batch: list[dict]) -> None:
-        try:
-            async with SessionLocal() as session:
-                for entry in batch:
-                    session.add(
-                        SecurityAccessLog(
-                            id=uuid.uuid4(),
-                            method=entry["method"],
+    async def _write_batch(self, batch: list[dict]) -> bool:
+        for attempt in range(WRITE_ATTEMPTS):
+            try:
+                async with SessionLocal() as session:
+                    for entry in batch:
+                        classification = classify_security_event(
                             path=entry["path"],
                             status_code=entry["status_code"],
-                            elapsed_ms=entry["elapsed_ms"],
-                            client_ip=entry.get("client_ip"),
-                            user_agent=entry.get("user_agent"),
-                            client_type=entry.get("client_type"),
-                            client_version=entry.get("client_version"),
-                            client_platform=entry.get("client_platform"),
-                            build_channel=entry.get("build_channel"),
-                            build_commit=entry.get("build_commit"),
                             auth_status=entry["auth_status"],
-                            user_identifier=entry.get("user_identifier"),
-                            created_at=entry.get("created_at", datetime.now(timezone.utc)),
+                            ip_location=resolve_ip_location(entry.get("client_ip")),
                         )
+                        session.add(
+                            SecurityAccessLog(
+                                id=uuid.uuid4(),
+                                method=entry["method"],
+                                path=entry["path"],
+                                status_code=entry["status_code"],
+                                elapsed_ms=entry["elapsed_ms"],
+                                client_ip=entry.get("client_ip"),
+                                user_agent=entry.get("user_agent"),
+                                client_type=entry.get("client_type"),
+                                client_version=entry.get("client_version"),
+                                client_platform=entry.get("client_platform"),
+                                build_channel=entry.get("build_channel"),
+                                build_commit=entry.get("build_commit"),
+                                request_headers=entry.get("request_headers"),
+                                request_snapshot=entry.get("request_snapshot"),
+                                request_id=entry.get("request_id"),
+                                category=classification["category"],
+                                severity=classification["severity"],
+                                auth_status=entry["auth_status"],
+                                user_identifier=entry.get("user_identifier"),
+                                created_at=entry.get("created_at", datetime.now(timezone.utc)),
+                            )
+                        )
+                    await session.commit()
+            except Exception as exc:
+                self._last_error_at = datetime.now(timezone.utc)
+                self._last_error_type = type(exc).__name__
+                if attempt + 1 < WRITE_ATTEMPTS:
+                    self._retry_count += 1
+                    logger.warning(
+                        "Security access log batch write failed; retrying (%d entries)",
+                        len(batch),
                     )
-                await session.commit()
-        except Exception:
-            logger.exception("Failed to write security access log batch (%d entries)", len(batch))
+                    await asyncio.sleep(0)
+                    continue
+                self._failed_batches += 1
+                self._failed_entries += len(batch)
+                logger.exception(
+                    "Failed to write security access log batch after retries (%d entries)",
+                    len(batch),
+                )
+                return False
+            self._written_entries += len(batch)
+            self._last_success_at = datetime.now(timezone.utc)
+            return True
+        return False
 
-    async def _drain(self) -> None:
-        batch: list[dict] = []
-        while not self._queue.empty():
-            try:
-                batch.append(self._queue.get_nowait())
-            except asyncio.QueueEmpty:
-                break
-        if batch:
-            await self._write_batch(batch)
+    def stats(self) -> dict[str, Any]:
+        return {
+            "running": bool(self._task and not self._task.done()),
+            "stopping": self._stopping,
+            "queue_size": self._queue.qsize(),
+            "queue_capacity": self._queue.maxsize,
+            "accepted_entries": self._accepted_entries,
+            "written_entries": self._written_entries,
+            "dropped_entries": self._dropped_entries,
+            "failed_batches": self._failed_batches,
+            "failed_entries": self._failed_entries,
+            "retry_count": self._retry_count,
+            "started_at": self._started_at.isoformat() if self._started_at else None,
+            "last_success_at": self._last_success_at.isoformat() if self._last_success_at else None,
+            "last_error_at": self._last_error_at.isoformat() if self._last_error_at else None,
+            "last_error_type": self._last_error_type,
+        }
 
 
 _writer: SecurityAccessLogWriter | None = None

backend/app/services/security_events.py

@@ -0,0 +1,37 @@
+"""Security event classification shared by writers and monitoring APIs."""
+
+from __future__ import annotations
+
+from app.services.ip_location import IpLocation
+
+
+SENSITIVE_PROBE_MARKERS = (
+    "/.env",
+    ".env",
+    "/.git",
+    ".git/config",
+    "backup",
+    "config.php",
+    "wp-config",
+    "database.yml",
+)
+
+def classify_security_event(
+    *,
+    path: str,
+    status_code: int,
+    auth_status: str,
+    ip_location: IpLocation | None = None,
+) -> dict[str, str]:
+    normalized_path = (path or "").lower()
+    if any(marker in normalized_path for marker in SENSITIVE_PROBE_MARKERS):
+        return {"category": "PROBE", "severity": "CRITICAL"}
+    if status_code >= 500:
+        return {"category": "SERVER_ERROR", "severity": "HIGH"}
+    if auth_status == "INVALID_TOKEN":
+        return {"category": "INVALID_TOKEN", "severity": "MEDIUM"}
+    if auth_status == "ANONYMOUS" and status_code in {401, 403}:
+        return {"category": "ANONYMOUS_API", "severity": "MEDIUM"}
+    if status_code == 404:
+        return {"category": "NOT_FOUND_NOISE", "severity": "LOW"}
+    return {"category": "OTHER", "severity": "LOW"}

backend/app/services/source_location_aggregator.py

@@ -0,0 +1,272 @@
+"""Hourly source-location aggregation and timeline reads."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import hmac
+import logging
+import uuid
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
+from typing import Literal
+
+from sqlalchemy import delete, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.permission_access_log import PermissionAccessLog
+from app.models.security_access_log import SecurityAccessLog
+from app.models.source_location_snapshot import SourceLocationSnapshot
+from app.services.geo_location_metadata import resolve_geo_location_metadata
+from app.services.ip_location import resolve_ip_location
+
+logger = logging.getLogger("ctms.source_location_aggregator")
+
+
+def _normalize_datetime(value: datetime) -> datetime:
+    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
+
+
+def _identity_hash(kind: str, value: str | None) -> str:
+    if not value:
+        return ""
+    payload = f"source-location:{kind}:{value}".encode("utf-8")
+    return hmac.new(settings.JWT_SECRET_KEY.encode("utf-8"), payload, hashlib.sha256).hexdigest()
+
+
+async def aggregate_source_location_hour(bucket_start: datetime, bucket_end: datetime) -> int:
+    bucket_start = _normalize_datetime(bucket_start)
+    bucket_end = _normalize_datetime(bucket_end)
+    async with SessionLocal() as session:
+        matching_security_request = select(SecurityAccessLog.id).where(
+            PermissionAccessLog.request_id.is_not(None),
+            SecurityAccessLog.request_id == PermissionAccessLog.request_id,
+        ).exists()
+        permission_rows = (
+            await session.execute(
+                select(
+                    PermissionAccessLog.ip_address,
+                    PermissionAccessLog.user_id,
+                    func.count().filter(PermissionAccessLog.allowed.is_(True)).label("allowed_count"),
+                    func.count().filter(PermissionAccessLog.allowed.is_(False)).label("denied_count"),
+                    func.min(PermissionAccessLog.created_at).label("first_seen_at"),
+                    func.max(PermissionAccessLog.created_at).label("last_seen_at"),
+                )
+                .where(
+                    PermissionAccessLog.created_at >= bucket_start,
+                    PermissionAccessLog.created_at < bucket_end,
+                    PermissionAccessLog.ip_address.is_not(None),
+                    ~matching_security_request,
+                )
+                .group_by(PermissionAccessLog.ip_address, PermissionAccessLog.user_id)
+            )
+        ).all()
+        security_rows = (
+            await session.execute(
+                select(
+                    SecurityAccessLog.client_ip,
+                    SecurityAccessLog.user_identifier,
+                    func.count().filter(SecurityAccessLog.status_code < 400).label("allowed_count"),
+                    func.count().filter(SecurityAccessLog.status_code >= 400).label("denied_count"),
+                    func.count()
+                    .filter(
+                        SecurityAccessLog.category.is_not(None),
+                        ~SecurityAccessLog.category.in_(("OTHER", "NOT_FOUND_NOISE")),
+                    )
+                    .label("security_event_count"),
+                    func.count()
+                    .filter(SecurityAccessLog.severity.in_(("HIGH", "CRITICAL")))
+                    .label("high_risk_count"),
+                    func.count()
+                    .filter(
+                        SecurityAccessLog.status_code >= 400,
+                        SecurityAccessLog.auth_status.in_(("INVALID_TOKEN", "ANONYMOUS")),
+                    )
+                    .label("auth_failure_count"),
+                    func.min(SecurityAccessLog.created_at).label("first_seen_at"),
+                    func.max(SecurityAccessLog.created_at).label("last_seen_at"),
+                )
+                .where(
+                    SecurityAccessLog.created_at >= bucket_start,
+                    SecurityAccessLog.created_at < bucket_end,
+                    SecurityAccessLog.client_ip.is_not(None),
+                )
+                .group_by(SecurityAccessLog.client_ip, SecurityAccessLog.user_identifier)
+            )
+        ).all()
+
+        aggregates: dict[tuple[str, str], dict] = {}
+
+        def merge_row(
+            ip_address: str,
+            user_identity: str,
+            allowed_count: int,
+            denied_count: int,
+            first_seen_at: datetime,
+            last_seen_at: datetime,
+            *,
+            security_event_count: int = 0,
+            high_risk_count: int = 0,
+            auth_failure_count: int = 0,
+        ) -> None:
+            key = (ip_address, user_identity)
+            row = aggregates.setdefault(
+                key,
+                {
+                    "ip_address": ip_address,
+                    "user_identity": user_identity,
+                    "allowed_count": 0,
+                    "denied_count": 0,
+                    "security_event_count": 0,
+                    "high_risk_count": 0,
+                    "auth_failure_count": 0,
+                    "first_seen_at": _normalize_datetime(first_seen_at),
+                    "last_seen_at": _normalize_datetime(last_seen_at),
+                },
+            )
+            row["allowed_count"] += int(allowed_count or 0)
+            row["denied_count"] += int(denied_count or 0)
+            row["security_event_count"] += int(security_event_count or 0)
+            row["high_risk_count"] += int(high_risk_count or 0)
+            row["auth_failure_count"] += int(auth_failure_count or 0)
+            row["first_seen_at"] = min(row["first_seen_at"], _normalize_datetime(first_seen_at))
+            row["last_seen_at"] = max(row["last_seen_at"], _normalize_datetime(last_seen_at))
+
+        for ip_address, user_id, allowed, denied, first_seen, last_seen in permission_rows:
+            merge_row(str(ip_address), str(user_id or ""), allowed, denied, first_seen, last_seen)
+        for ip_address, user_identifier, allowed, denied, security_events, high_risk, auth_failures, first_seen, last_seen in security_rows:
+            merge_row(
+                str(ip_address),
+                str(user_identifier or ""),
+                allowed,
+                denied,
+                first_seen,
+                last_seen,
+                security_event_count=security_events,
+                high_risk_count=high_risk,
+                auth_failure_count=auth_failures,
+            )
+
+        await session.execute(
+            delete(SourceLocationSnapshot).where(SourceLocationSnapshot.bucket_time == bucket_start)
+        )
+        for row in aggregates.values():
+            ip_info = resolve_ip_location(row["ip_address"])
+            metadata = resolve_geo_location_metadata(ip_info)
+            longitude = metadata.longitude
+            latitude = metadata.latitude
+            country = metadata.country or ip_info.country
+            location = " / ".join(
+                part for part in [country, ip_info.province, ip_info.city] if part
+            ) or ip_info.location or "未知"
+            session.add(
+                SourceLocationSnapshot(
+                    id=uuid.uuid4(),
+                    bucket_time=bucket_start,
+                    ip_hash=_identity_hash("ip", row["ip_address"]),
+                    user_hash=_identity_hash("user", row["user_identity"]),
+                    country=country,
+                    country_code=metadata.country_code,
+                    province=ip_info.province,
+                    region_code=metadata.region_code,
+                    city=ip_info.city,
+                    isp=ip_info.isp,
+                    location=location,
+                    longitude=longitude,
+                    latitude=latitude,
+                    accuracy_level=metadata.accuracy_level,
+                    allowed_count=row["allowed_count"],
+                    denied_count=row["denied_count"],
+                    security_event_count=row["security_event_count"],
+                    high_risk_count=row["high_risk_count"],
+                    auth_failure_count=row["auth_failure_count"],
+                    first_seen_at=row["first_seen_at"],
+                    last_seen_at=row["last_seen_at"],
+                )
+            )
+        await session.commit()
+        return len(aggregates)
+
+
+async def get_source_location_timeline(
+    db: AsyncSession,
+    *,
+    start_at: datetime,
+    end_at: datetime,
+    granularity: Literal["hour", "day"],
+) -> list[dict]:
+    rows = (
+        await db.execute(
+            select(SourceLocationSnapshot)
+            .where(
+                SourceLocationSnapshot.bucket_time >= start_at,
+                SourceLocationSnapshot.bucket_time < end_at,
+            )
+            .order_by(SourceLocationSnapshot.bucket_time)
+        )
+    ).scalars().all()
+    buckets: dict[datetime, dict] = defaultdict(
+        lambda: {
+            "allowed_count": 0,
+            "denied_count": 0,
+            "security_event_count": 0,
+            "high_risk_count": 0,
+            "ip_hashes": set(),
+            "user_hashes": set(),
+        }
+    )
+    for row in rows:
+        bucket = _normalize_datetime(row.bucket_time)
+        if granularity == "day":
+            bucket = bucket.replace(hour=0, minute=0, second=0, microsecond=0)
+        else:
+            bucket = bucket.replace(minute=0, second=0, microsecond=0)
+        item = buckets[bucket]
+        item["allowed_count"] += row.allowed_count
+        item["denied_count"] += row.denied_count
+        item["security_event_count"] += row.security_event_count
+        item["high_risk_count"] += row.high_risk_count
+        item["ip_hashes"].add(row.ip_hash)
+        if row.user_hash:
+            item["user_hashes"].add(row.user_hash)
+
+    return [
+        {
+            "bucket_time": bucket.isoformat(),
+            "total_count": item["allowed_count"] + item["denied_count"],
+            "allowed_count": item["allowed_count"],
+            "denied_count": item["denied_count"],
+            "security_event_count": item["security_event_count"],
+            "high_risk_count": item["high_risk_count"],
+            "unique_ip_count": len(item["ip_hashes"]),
+            "unique_user_count": len(item["user_hashes"]),
+        }
+        for bucket, item in sorted(buckets.items())
+    ]
+
+
+async def run_hourly_source_location_aggregation(stop_event: asyncio.Event) -> None:
+    logger.info("Source location aggregator started")
+    now = datetime.now(timezone.utc)
+    completed_hour = now.replace(minute=0, second=0, microsecond=0)
+    try:
+        await aggregate_source_location_hour(completed_hour - timedelta(hours=1), completed_hour)
+    except Exception:
+        logger.exception("Failed to backfill source location snapshot for %s", completed_hour)
+
+    while not stop_event.is_set():
+        now = datetime.now(timezone.utc)
+        next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
+        try:
+            await asyncio.wait_for(stop_event.wait(), timeout=(next_hour - now).total_seconds())
+            break
+        except asyncio.TimeoutError:
+            pass
+        try:
+            await aggregate_source_location_hour(next_hour - timedelta(hours=1), next_hour)
+        except Exception:
+            logger.exception("Failed to aggregate source locations for %s", next_hour)
+
+    logger.info("Source location aggregator stopped")

backend/app/services/user_login_sessions.py

@@ -0,0 +1,198 @@
+"""Server-authoritative login session activity for the admin account list."""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any, Iterable
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.models.user_login_session import UserLoginSession
+
+
+@dataclass(frozen=True)
+class UserLoginSummary:
+    status: str = "OFFLINE"
+    last_login_at: datetime | None = None
+    last_seen_at: datetime | None = None
+    client_type: str | None = None
+    active_session_count: int = 0
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _as_utc(value: datetime) -> datetime:
+    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
+
+
+def _text_header(headers: Any, name: str, limit: int) -> str | None:
+    value = (headers.get(name) or "").strip()
+    return value[:limit] or None
+
+
+def session_id_from_payload(payload: dict[str, Any]) -> uuid.UUID:
+    raw_session_id = payload.get("sid")
+    try:
+        return uuid.UUID(str(raw_session_id))
+    except (TypeError, ValueError):
+        # Tokens issued before session tracking are mapped to a deterministic
+        # legacy session without persisting the token or a token-derived value.
+        legacy_key = ":".join(
+            [
+                str(payload.get("sub") or ""),
+                str(payload.get("orig_iat") or payload.get("iat") or ""),
+                str(payload.get("client_type") or "web"),
+            ]
+        )
+        return uuid.uuid5(uuid.NAMESPACE_URL, f"ctms:legacy-session:{legacy_key}")
+
+
+def session_client_type(payload: dict[str, Any], headers: Any) -> str:
+    from_payload = (payload.get("client_type") or "").strip().lower()
+    from_header = (headers.get("x-ctms-client-type") or "").strip().lower()
+    return "desktop" if "desktop" in {from_payload, from_header} else "web"
+
+
+async def create_login_session(
+    db: AsyncSession,
+    *,
+    session_id: uuid.UUID,
+    user_id: uuid.UUID,
+    request: Any,
+    login_at: datetime | None = None,
+) -> UserLoginSession:
+    occurred_at = login_at or _now()
+    session = UserLoginSession(
+        id=session_id,
+        user_id=user_id,
+        client_type=session_client_type({}, request.headers),
+        client_platform=_text_header(request.headers, "x-ctms-client-platform", 32),
+        client_version=_text_header(request.headers, "x-ctms-client-version", 64),
+        client_source=_text_header(request.headers, "x-ctms-client-source", 32),
+        login_at=occurred_at,
+        last_seen_at=occurred_at,
+    )
+    db.add(session)
+    await db.commit()
+    await db.refresh(session)
+    return session
+
+
+async def touch_login_session(
+    db: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    payload: dict[str, Any],
+    request: Any,
+) -> UserLoginSession | None:
+    session_id = session_id_from_payload(payload)
+    session = await db.get(UserLoginSession, session_id)
+    now = _now()
+    if session is None:
+        session = UserLoginSession(
+            id=session_id,
+            user_id=user_id,
+            client_type=session_client_type(payload, request.headers),
+            client_platform=_text_header(request.headers, "x-ctms-client-platform", 32),
+            client_version=_text_header(request.headers, "x-ctms-client-version", 64),
+            client_source=_text_header(request.headers, "x-ctms-client-source", 32),
+            login_at=now,
+            last_seen_at=now,
+        )
+        db.add(session)
+    elif session.user_id != user_id or session.ended_at is not None:
+        return None
+    else:
+        session.last_seen_at = now
+        session.client_platform = _text_header(request.headers, "x-ctms-client-platform", 32) or session.client_platform
+        session.client_version = _text_header(request.headers, "x-ctms-client-version", 64) or session.client_version
+    await db.commit()
+    await db.refresh(session)
+    return session
+
+
+async def end_login_session(
+    db: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    payload: dict[str, Any],
+    reason: str = "logout",
+) -> bool:
+    session_id = session_id_from_payload(payload)
+    result = await db.execute(
+        update(UserLoginSession)
+        .where(
+            UserLoginSession.id == session_id,
+            UserLoginSession.user_id == user_id,
+            UserLoginSession.ended_at.is_(None),
+        )
+        .values(ended_at=_now(), end_reason=reason, last_seen_at=_now())
+    )
+    await db.commit()
+    return bool(result.rowcount)
+
+
+async def get_login_summaries(
+    db: AsyncSession,
+    user_ids: Iterable[uuid.UUID],
+) -> dict[uuid.UUID, UserLoginSummary]:
+    ids = list(user_ids)
+    if not ids:
+        return {}
+    rows = (
+        await db.execute(
+            select(UserLoginSession)
+            .where(UserLoginSession.user_id.in_(ids))
+            .order_by(UserLoginSession.user_id, UserLoginSession.login_at.desc())
+        )
+    ).scalars().all()
+    cutoff = _now() - timedelta(seconds=settings.USER_SESSION_ONLINE_SECONDS)
+    summaries: dict[uuid.UUID, UserLoginSummary] = {}
+    mutable: dict[uuid.UUID, dict[str, Any]] = {}
+    for row in rows:
+        current = mutable.setdefault(
+            row.user_id,
+            {
+                "last_login_at": row.login_at,
+                "last_seen_at": row.last_seen_at,
+                "client_type": row.client_type,
+                "active_session_count": 0,
+            },
+        )
+        row_last_seen_at = _as_utc(row.last_seen_at)
+        if row_last_seen_at and (
+            current["last_seen_at"] is None or row_last_seen_at > _as_utc(current["last_seen_at"])
+        ):
+            current["last_seen_at"] = row_last_seen_at
+        if row.ended_at is None and row_last_seen_at >= cutoff:
+            current["active_session_count"] += 1
+    for user_id, item in mutable.items():
+        summaries[user_id] = UserLoginSummary(
+            status="ONLINE" if item["active_session_count"] else "OFFLINE",
+            last_login_at=item["last_login_at"],
+            last_seen_at=item["last_seen_at"],
+            client_type=item["client_type"],
+            active_session_count=item["active_session_count"],
+        )
+    return summaries
+
+
+async def list_login_activities(
+    db: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    limit: int,
+) -> list[UserLoginSession]:
+    result = await db.execute(
+        select(UserLoginSession)
+        .where(UserLoginSession.user_id == user_id)
+        .order_by(UserLoginSession.login_at.desc())
+        .limit(limit)
+    )
+    return list(result.scalars().all())