perf(analytics): 5表 updated_at 窄索引——health MAX 消除全表扫(128MB buffer pool)

Co-Authored-By: Claude Code <noreply@anthropic.com>

yunke-server/app/db/models.py

@@ -256,6 +256,9 @@ class Customer(Base):
             "owner_account_id",
             "friend_id",
         ),
+        # /api/analytics/health MAX(updated_at) 新鲜度探针;无索引时全表扫。
+        # 「模型+迁移双声明」防 alembic autogenerate 反 DROP(c3d4e5f6a7b8)。
+        Index("ix_customer_updated_at", "updated_at"),
     )
 
 
@@ -314,6 +317,8 @@ class ChatGroup(Base):
             "owner_account_id",
             "room_id",
         ),
+        # /api/analytics/health MAX(updated_at) 新鲜度探针(c3d4e5f6a7b8)。
+        Index("ix_chat_group_updated_at", "updated_at"),
     )
 
 
@@ -480,6 +485,9 @@ class Media(Base):
 
     __table_args__ = (
         Index("ix_media_type_channel", "media_type", "channel"),
+        # /api/analytics/health MAX(updated_at) 新鲜度探针——media 5 万宽行
+        # (多个 TEXT/JSON 列)在 128MB buffer pool 下最伤(c3d4e5f6a7b8)。
+        Index("ix_media_updated_at", "updated_at"),
     )
 
 
@@ -531,6 +539,8 @@ class PhoneCall(Base):
 
     __table_args__ = (
         Index("ix_phone_call_user_time", "user_id", "start_call_time"),
+        # /api/analytics/health MAX(updated_at) 新鲜度探针(c3d4e5f6a7b8)。
+        Index("ix_phone_call_updated_at", "updated_at"),
     )
 
 
@@ -649,6 +659,9 @@ class MomentsPost(Base):
     fetched_at: Mapped[dt.datetime] = _fetched_at()
     updated_at: Mapped[dt.datetime] = _updated_at()
 
+    # /api/analytics/health MAX(updated_at) 新鲜度探针(c3d4e5f6a7b8)。
+    __table_args__ = (Index("ix_moments_post_updated_at", "updated_at"),)
+
 
 # ===========================================================================
 # 5. Operational / bookkeeping tables (local; no raw_json/version)

yunke-server/migrations/versions/c3d4e5f6a7b8_freshness_indexes.py

@@ -0,0 +1,50 @@
+"""5 表 updated_at 窄索引——/api/analytics/health MAX 消除全表扫
+
+Revision ID: c3d4e5f6a7b8
+Revises: f0a1b2c3d4e5
+Create Date: 2026-09-04
+
+health 端点每页面加载对 6 表各 MAX 一次;5 表 updated_at 无索引时
+MySQL 全表扫(media 5万宽行在 128MB buffer pool 下最伤)。单列窄索引
+各几 MB,MAX 变 O(1) 索引定位。幂等:_has_index 先查。SQLite/MySQL 通用。
+模型侧 models.py 同步双声明,防 alembic autogenerate 反 DROP。
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "c3d4e5f6a7b8"
+down_revision: str | Sequence[str] | None = "f0a1b2c3d4e5"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+# (表名, 索引名) —— 与 models.py 各模型 __table_args__ 一字对应
+_INDEXES: tuple[tuple[str, str], ...] = (
+    ("customer", "ix_customer_updated_at"),
+    ("chat_group", "ix_chat_group_updated_at"),
+    ("phone_call", "ix_phone_call_updated_at"),
+    ("media", "ix_media_updated_at"),
+    ("moments_post", "ix_moments_post_updated_at"),
+)
+
+
+def _has_index(bind, table: str, index: str) -> bool:
+    insp = sa.inspect(bind)
+    if not insp.has_table(table):
+        return False
+    return index in {ix["name"] for ix in insp.get_indexes(table)}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    for table, index in _INDEXES:
+        if not _has_index(bind, table, index):
+            op.create_index(index, table, ["updated_at"])
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    for table, index in _INDEXES:
+        if _has_index(bind, table, index):
+            op.drop_index(index, table_name=table)

yunke-server/tests/test_freshness_indexes.py

@@ -0,0 +1,42 @@
+"""5 个新鲜度索引:create_all 建出 + 模型迁移双声明一致(防 autogenerate 反 DROP)。
+
+迁移侧(c3d4e5f6a7b8)由临时库 alembic upgrade head 验证;本文件锁定
+模型侧——Base.metadata 与 models.py __table_args__ 声明一旦被误删,
+create_all 建出的表就缺索引,此测试即红。
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.db.base import Base
+
+EXPECTED = {
+    "customer": "ix_customer_updated_at",
+    "chat_group": "ix_chat_group_updated_at",
+    "phone_call": "ix_phone_call_updated_at",
+    "media": "ix_media_updated_at",
+    "moments_post": "ix_moments_post_updated_at",
+}
+
+
+async def test_freshness_indexes_exist():
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+            names = await conn.run_sync(
+                lambda sync_conn: {
+                    t: {ix["name"] for ix in sa.inspect(sync_conn).get_indexes(t)}
+                    for t in EXPECTED
+                }
+            )
+        for table, index in EXPECTED.items():
+            assert index in names[table], f"{table}.{index} missing"
+    finally:
+        await engine.dispose()