feat(ci,alembic): Hub Docker镜像构建发布，数据库迁移并发安全加固

- build-hub.yml 新增 Docker job，构建多架构镜像推送至 GHCR 和 Docker Hub
- build-hub/build-proxy Release 名称简化为 tag 名
- alembic/env.py 使用 PostgreSQL advisory lock 防止多进程并发迁移竞态
- 迁移脚本改用 ADD/DROP COLUMN IF NOT EXISTS 替代 inspector 检查

alembic/env.py

@@ -3,13 +3,15 @@ Alembic 环境配置
 用于数据库迁移的运行时环境设置
 """
 
-from logging.config import fileConfig
-from sqlalchemy import engine_from_config, pool
-from alembic import context
 import os
 import sys
+from logging.config import fileConfig
 from pathlib import Path
 
+from sqlalchemy import engine_from_config, pool, text
+
+from alembic import context
+
 # 添加项目根目录到 Python 路径
 sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
 
@@ -48,6 +50,10 @@ if config.config_file_name is not None:
 # 目标元数据（包含所有表定义）
 target_metadata = Base.metadata
 
+# PostgreSQL 全局迁移锁，避免多进程并发执行 Alembic 导致竞态（重复加列/索引等）
+# ID 由 crc32("aether-alembic-migration") 拼接生成，仅需全局唯一即可
+MIGRATION_ADVISORY_LOCK_ID = 582694137405821
+
 
 def run_migrations_offline() -> None:
     """
@@ -83,15 +89,31 @@ def run_migrations_online() -> None:
     )
 
     with connectable.connect() as connection:
-        context.configure(
-            connection=connection,
-            target_metadata=target_metadata,
-            compare_type=True,  # 比较列类型变更
-            compare_server_default=True,  # 比较默认值变更
-        )
+        lock_acquired = False
+        try:
+            if connection.dialect.name == "postgresql":
+                connection.execute(
+                    text("SELECT pg_advisory_lock(:lock_id)"),
+                    {"lock_id": MIGRATION_ADVISORY_LOCK_ID},
+                )
+                lock_acquired = True
 
-        with context.begin_transaction():
-            context.run_migrations()
+            context.configure(
+                connection=connection,
+                target_metadata=target_metadata,
+                compare_type=True,  # 比较列类型变更
+                compare_server_default=True,  # 比较默认值变更
+            )
+
+            with context.begin_transaction():
+                context.run_migrations()
+        finally:
+            if lock_acquired:
+                connection.rollback()
+                connection.execute(
+                    text("SELECT pg_advisory_unlock(:lock_id)"),
+                    {"lock_id": MIGRATION_ADVISORY_LOCK_ID},
+                )
 
 
 # 根据模式选择运行方式

alembic/versions/20260220_1800_7e8f9a0b1c2d_add_provider_request_body_and_client_response_body.py

@@ -10,8 +10,7 @@ from __future__ import annotations
 
 from collections.abc import Sequence
 
-import sqlalchemy as sa
-from sqlalchemy import inspect
+from sqlalchemy import text
 
 from alembic import op
 
@@ -24,33 +23,24 @@ depends_on: str | Sequence[str] | None = None
 
 def upgrade() -> None:
     conn = op.get_bind()
-    inspector = inspect(conn)
-    existing_columns = {col["name"] for col in inspector.get_columns("usage")}
-
-    if "provider_request_body" not in existing_columns:
-        op.add_column("usage", sa.Column("provider_request_body", sa.JSON(), nullable=True))
-    if "provider_request_body_compressed" not in existing_columns:
-        op.add_column(
-            "usage", sa.Column("provider_request_body_compressed", sa.LargeBinary(), nullable=True)
-        )
-    if "client_response_body" not in existing_columns:
-        op.add_column("usage", sa.Column("client_response_body", sa.JSON(), nullable=True))
-    if "client_response_body_compressed" not in existing_columns:
-        op.add_column(
-            "usage", sa.Column("client_response_body_compressed", sa.LargeBinary(), nullable=True)
-        )
+    # Use PostgreSQL native IF NOT EXISTS to avoid duplicate-column races
+    # when migrations are triggered concurrently (e.g. startup + manual run).
+    conn.execute(text("ALTER TABLE usage ADD COLUMN IF NOT EXISTS provider_request_body JSON"))
+    conn.execute(
+        text("ALTER TABLE usage ADD COLUMN IF NOT EXISTS provider_request_body_compressed BYTEA")
+    )
+    conn.execute(text("ALTER TABLE usage ADD COLUMN IF NOT EXISTS client_response_body JSON"))
+    conn.execute(
+        text("ALTER TABLE usage ADD COLUMN IF NOT EXISTS client_response_body_compressed BYTEA")
+    )
 
 
 def downgrade() -> None:
     conn = op.get_bind()
-    inspector = inspect(conn)
-    existing_columns = {col["name"] for col in inspector.get_columns("usage")}
-
     for col in (
         "client_response_body_compressed",
         "client_response_body",
         "provider_request_body_compressed",
         "provider_request_body",
     ):
-        if col in existing_columns:
-            op.drop_column("usage", col)
+        conn.execute(text(f"ALTER TABLE usage DROP COLUMN IF EXISTS {col}"))