feat: add supabase

.gitignore

@@ -88,6 +88,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+.env.local
 
 .spyderproject
 .spyproject
@@ -131,4 +132,6 @@ test_vector_store_async.py
 deploy_sdk.py
 tests
 deploy_hf.py
-.mimocode
+.mimocode
+
+ddl

app/api/server.py

@@ -15,7 +15,6 @@ from app.core.database import pool_manager
 from app.core.logger import get_logger
 from app.core.redis_client import close_redis, create_redis_client
 from app.core.scripts import load_scripts
-from app.core.tidb_manager import TiDBWriteError
 from app.services.embeddings_service import EmbeddingService
 from app.services.vector_store_service import VectorStoreService
 
@@ -44,18 +43,15 @@ async def _self_ping():
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    if not _settings.tidb_instance_list:
+    if not _settings.supabase_url or not _settings.supabase_service_role_key:
         _logger.error(
-            "No TiDB instances configured! Set TIDB_INSTANCES environment variable "
-            "with comma-separated MySQL connection strings."
+            "Supabase not configured! Set SUPABASE_URL and "
+            "SUPABASE_SERVICE_ROLE_KEY environment variables."
         )
     else:
-        _logger.info(
-            "Initializing TiDB databases across %d instance(s)...",
-            len(_settings.tidb_instance_list),
-        )
+        _logger.info("Initializing Supabase databases...")
         await init_auth_db()
-        _logger.info("Authentication database initialized on all TiDB instances")
+        _logger.info("Authentication database initialized via Supabase")
 
         _logger.info("Initializing vector store database...")
         from app.core.vector_store.deps import init_vector_store_db
@@ -87,11 +83,11 @@ async def lifespan(app: FastAPI):
     await close_redis(redis)
     await _vector_store_service.close_all()
     await pool_manager.close_all()
-    from app.core.tidb_manager import get_tidb_manager
-    mgr = get_tidb_manager()
-    if mgr:
-        await mgr.close_all_pools()
-        _logger.info("TiDB connection pools closed")
+    from app.services.supabase import get_supabase_client
+    client = get_supabase_client()
+    if client:
+        await client.close()
+        _logger.info("Supabase client closed")
 
 
 def create_application() -> FastAPI:
@@ -121,18 +117,6 @@ def create_application() -> FastAPI:
         allow_headers=["*"],
     )
 
-    @app.exception_handler(TiDBWriteError)
-    async def tidb_write_error_handler(request: Request, exc: TiDBWriteError):
-        from starlette.responses import JSONResponse
-        return JSONResponse(
-            status_code=503,
-            content={
-                "success": False,
-                "detail": "Failed to process request due to database storage limit. "
-                          "Please try again later or contact support.",
-            },
-        )
-
     @app.middleware("http")
     async def maintenance_middleware(request: Request, call_next):
         if is_maintenance():

app/api/v1/auth.py

@@ -5,7 +5,7 @@ from typing import Annotated
 from fastapi import APIRouter, Depends, Request, status
 
 from app.config import get_settings
-from app.core.auth.deps import get_current_user, get_db, require_application_id
+from app.core.auth.deps import get_current_user, get_db
 from app.core.auth.models import User
 from app.core.auth.schemas import (
     ChangePasswordSchema,
@@ -24,140 +24,120 @@ from app.core.auth.schemas import (
     UserSchemaField,
     UserSchemaResponse,
 )
-from app.core.tidb_manager import TiDBManager
+from app.services.supabase import SupabaseClient
 from app.services.auth_service import AuthService
 
 router = APIRouter(prefix="/auth", tags=["Authentication"])
 _settings = get_settings()
 
 
-def _build(data, application_id=None):
-    result = {"success": True, "data": data}
-    if application_id:
-        result["application_id"] = application_id
-    return result
-
-
 @router.post("/register", response_model=ProfileResponse, status_code=status.HTTP_201_CREATED)
 async def register(
     schema: RegisterSchema,
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     user = await AuthService.register(db, schema)
     profile = await AuthService.user_to_profile(db, user)
-    return _build(profile.model_dump(), _settings.application_id or None)
+    return {"success": True, "data": profile.model_dump()}
 
 
 @router.post("/login", response_model=TokenResponse)
 async def login(
     schema: LoginSchema,
     request: Request,
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     tokens = await AuthService.login(db, request, schema)
-    return _build(tokens.model_dump(), _settings.application_id or None)
+    return {"success": True, "data": tokens.model_dump()}
 
 
 @router.post("/refresh", response_model=TokenResponse)
 async def refresh_token(
     schema: TokenRefreshSchema,
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     tokens = await AuthService.refresh(db, schema.refresh_token)
-    return _build(tokens.model_dump(), _settings.application_id or None)
+    return {"success": True, "data": tokens.model_dump()}
 
 
 @router.post("/logout", response_model=MessageDataResponse)
 async def logout(
     schema: TokenRefreshSchema,
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     await AuthService.logout(db, current_user, schema.refresh_token)
-    return _build({"message": "Logged out successfully"}, _settings.application_id or None)
+    return {"success": True, "data": {"message": "Logged out successfully"}}
 
 
 @router.post("/logout-all", response_model=MessageDataResponse)
 async def logout_all(
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     await AuthService.logout_all(db, current_user)
-    return _build({"message": "All sessions revoked"}, _settings.application_id or None)
+    return {"success": True, "data": {"message": "All sessions revoked"}}
 
 
 @router.post("/forgot-password", response_model=MessageDataResponse)
 async def forgot_password(
     schema: ForgotPasswordSchema,
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     await AuthService.forgot_password(db, schema)
-    return _build({"message": "If that email exists, a password reset link has been sent."}, _settings.application_id or None)
+    return {"success": True, "data": {"message": "If that email exists, a password reset link has been sent."}}
 
 
 @router.post("/reset-password", response_model=MessageDataResponse)
 async def reset_password(
     schema: ResetPasswordSchema,
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     await AuthService.reset_password(db, schema)
-    return _build({"message": "Password reset successfully"}, _settings.application_id or None)
+    return {"success": True, "data": {"message": "Password reset successfully"}}
 
 
 @router.post("/change-password", response_model=MessageDataResponse)
 async def change_password(
     schema: ChangePasswordSchema,
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     await AuthService.change_password(db, current_user, schema)
-    return _build({"message": "Password changed successfully"}, _settings.application_id or None)
+    return {"success": True, "data": {"message": "Password changed successfully"}}
 
 
 @router.get("/me", response_model=ProfileResponse)
 async def get_me(
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     profile = await AuthService.user_to_profile(db, current_user)
-    return _build(profile.model_dump(), _settings.application_id or None)
+    return {"success": True, "data": profile.model_dump()}
 
 
 @router.patch("/me", response_model=ProfileResponse)
 async def update_me(
     schema: UpdateProfileSchema,
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     user = await AuthService.update_profile(db, current_user, schema)
     profile = await AuthService.user_to_profile(db, user)
-    return _build(profile.model_dump(), _settings.application_id or None)
+    return {"success": True, "data": profile.model_dump()}
 
 
 @router.delete("/me", response_model=MessageDataResponse)
 async def delete_me(
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     await AuthService.soft_delete(db, current_user)
-    return _build({"message": "Account deleted successfully"}, _settings.application_id or None)
+    return {"success": True, "data": {"message": "Account deleted successfully"}}
 
 
 @router.get("/schema", response_model=SchemaResponse)
-async def get_user_schema(
-    _: Annotated[bool, Depends(require_application_id)],
-):
+async def get_user_schema():
     columns = [
         UserSchemaField(field="id", type="string (UUID)", required=True, description="Unique user identifier", constraints="Auto-generated"),
         UserSchemaField(field="email", type="string", required=True, description="User email address", constraints="Unique, max 255 chars"),
@@ -176,25 +156,23 @@ async def get_user_schema(
         UserSchemaField(field="deleted_at", type="datetime (ISO 8601)", required=False, description="Soft delete timestamp", constraints="Nullable"),
         UserSchemaField(field="roles", type="array[string]", required=False, description="Assigned role names", constraints="Via user_roles association table"),
     ]
-    return _build(UserSchemaResponse(table_name="users", columns=columns).model_dump(), _settings.application_id or None)
+    return {"success": True, "data": UserSchemaResponse(table_name="users", columns=columns).model_dump()}
 
 
 @router.get("/sessions", response_model=SessionListResponse)
 async def list_sessions(
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     sessions = await AuthService.list_sessions(db, current_user)
-    return _build([SessionOut.model_validate(s).model_dump() for s in sessions], _settings.application_id or None)
+    return {"success": True, "data": [SessionOut.model_validate(s).model_dump() for s in sessions]}
 
 
 @router.delete("/sessions/{session_id}", response_model=MessageDataResponse)
 async def revoke_session(
     session_id: str,
     current_user: Annotated[User, Depends(get_current_user)],
-    db: Annotated[TiDBManager, Depends(get_db)],
-    _: Annotated[bool, Depends(require_application_id)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ):
     await AuthService.revoke_session(db, current_user, session_id)
-    return _build({"message": "Session revoked successfully"}, _settings.application_id or None)
+    return {"success": True, "data": {"message": "Session revoked successfully"}}

app/api/v1/url_shortener.py

@@ -36,10 +36,6 @@ def _get_service() -> URLShortenerService:
     return _shared_service
 
 
-def _app_id() -> str:
-    return _settings.application_id or ""
-
-
 # ---------------------------------------------------------------------------
 # Request / Response models
 # ---------------------------------------------------------------------------
@@ -54,7 +50,6 @@ class CreateOwnerResponse(BaseModel):
     owner_id: str
     api_key: str
     plan: str
-    application_id: str = ""
 
 
 class ShortenRequest(BaseModel):
@@ -94,7 +89,6 @@ class ShortLinkResponse(BaseModel):
     custom_domain: Optional[str] = None
     fallback_url: Optional[str] = None
     webhook_url: Optional[str] = None
-    application_id: str = ""
 
 
 class UpdateLinkRequest(BaseModel):
@@ -113,7 +107,6 @@ class UpdateLinkRequest(BaseModel):
 class ListLinksResponse(BaseModel):
     success: bool = True
     links: List[ShortLinkResponse]
-    application_id: str = ""
 
 
 class StatsResponse(BaseModel):
@@ -133,7 +126,6 @@ class StatsResponse(BaseModel):
     custom_domain: Optional[str] = None
     fallback_url: Optional[str] = None
     webhook_url: Optional[str] = None
-    application_id: str = ""
 
 
 class ClickDetail(BaseModel):
@@ -150,7 +142,6 @@ class ClickDetail(BaseModel):
 class ClickDetailsResponse(BaseModel):
     success: bool = True
     clicks: List[ClickDetail]
-    application_id: str = ""
 
 
 class QRCodeResponse(BaseModel):
@@ -158,7 +149,6 @@ class QRCodeResponse(BaseModel):
     short_code: str
     short_url: str
     qr_image_url: str
-    application_id: str = ""
 
 
 class ErrorResponse(BaseModel):
@@ -173,7 +163,6 @@ class BulkShortenRequest(BaseModel):
 class BulkShortenResponse(BaseModel):
     success: bool = True
     results: List[Dict[str, Any]]
-    application_id: str = ""
 
 
 class OwnerSummaryResponse(BaseModel):
@@ -185,7 +174,6 @@ class OwnerSummaryResponse(BaseModel):
     active_links: int
     total_clicks: int
     created_at: str
-    application_id: str = ""
 
 
 class CampaignCreateRequest(BaseModel):
@@ -203,13 +191,11 @@ class CampaignResponse(BaseModel):
     link_count: Optional[int] = 0
     total_clicks: Optional[int] = 0
     analytics: Optional[Dict[str, Any]] = None
-    application_id: str = ""
 
 
 class CampaignListResponse(BaseModel):
     success: bool = True
     campaigns: List[CampaignResponse]
-    application_id: str = ""
 
 
 class HealthCheckResponse(BaseModel):
@@ -219,7 +205,6 @@ class HealthCheckResponse(BaseModel):
     reachable: bool
     status_code: Optional[int] = None
     error: Optional[str] = None
-    application_id: str = ""
 
 
 # ---------------------------------------------------------------------------
@@ -251,7 +236,6 @@ def _build_link_response(link, service) -> ShortLinkResponse:
         custom_domain=link.custom_domain,
         fallback_url=link.fallback_url,
         webhook_url=link.webhook_url,
-        application_id=_app_id(),
     )
 
 
@@ -271,7 +255,6 @@ def create_owner(body: CreateOwnerRequest):
         owner_id=result["owner_id"],
         api_key=result["api_key"],
         plan=body.plan or "free",
-        application_id=_app_id(),
     )
 
 
@@ -283,7 +266,7 @@ def update_plan(plan: str = Query(..., description="free, pro, or enterprise"),
                 owner_id: str = Depends(_auth_owner)):
     service = _get_service()
     try:
-        return {"success": True, "application_id": _app_id(), **service.update_owner_plan(owner_id, plan)}
+        return {"success": True, **service.update_owner_plan(owner_id, plan)}
     except ValidationError as e:
         raise HTTPException(status_code=400, detail={"success": False, "error": str(e)})
 
@@ -298,7 +281,7 @@ def owner_summary(owner_id: str = Depends(_auth_owner)):
     data = service.owner_summary(owner_id)
     if not data:
         raise HTTPException(status_code=404, detail={"success": False, "error": "Owner not found"})
-    return OwnerSummaryResponse(application_id=_app_id(), **data)
+    return OwnerSummaryResponse(**data)
 
 
 # ---------------------------------------------------------------------------
@@ -361,7 +344,7 @@ def bulk_shorten(body: BulkShortenRequest, owner_id: str = Depends(_auth_owner))
         raise HTTPException(status_code=400, detail={"success": False, "error": "Bulk limit is 100 items per request"})
     service = _get_service()
     results = service.bulk_shorten(owner_id, body.items)
-    return BulkShortenResponse(results=results, application_id=_app_id())
+    return BulkShortenResponse(results=results, )
 
 
 @router.get(
@@ -373,7 +356,7 @@ def list_links(owner_id: str = Depends(_auth_owner)):
     service = _get_service()
     links = service.list_links(owner_id)
     items = [_build_link_response(link, service) for link in links]
-    return ListLinksResponse(links=items, application_id=_app_id())
+    return ListLinksResponse(links=items, )
 
 
 @router.get(
@@ -400,7 +383,6 @@ def create_campaign(body: CampaignCreateRequest, owner_id: str = Depends(_auth_o
             campaign_id=result["campaign_id"],
             name=result["name"],
             description=result.get("description"),
-            application_id=_app_id(),
         )
     except PlanLimitExceededError as e:
         raise HTTPException(status_code=402, detail={"success": False, "error": str(e), "code": "plan_limit"})
@@ -416,8 +398,8 @@ def create_campaign(body: CampaignCreateRequest, owner_id: str = Depends(_auth_o
 def list_campaigns(owner_id: str = Depends(_auth_owner)):
     service = _get_service()
     campaigns = service.list_campaigns(owner_id)
-    items = [CampaignResponse(**c, application_id=_app_id()) for c in campaigns]
-    return CampaignListResponse(campaigns=items, application_id=_app_id())
+    items = [CampaignResponse(**c, ) for c in campaigns]
+    return CampaignListResponse(campaigns=items, )
 
 
 @router.get(
@@ -432,7 +414,7 @@ def get_campaign(
     service = _get_service()
     try:
         data = service.get_campaign(campaign_id, owner_id)
-        return CampaignResponse(**data, application_id=_app_id())
+        return CampaignResponse(**data, )
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
     except AuthorizationError as e:
@@ -452,7 +434,7 @@ def list_campaign_links(
     try:
         links = service.list_campaign_links(campaign_id, owner_id)
         items = [_build_link_response(link, service) for link in links]
-        return ListLinksResponse(links=items, application_id=_app_id())
+        return ListLinksResponse(links=items, )
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
     except AuthorizationError as e:
@@ -470,7 +452,7 @@ def deactivate_campaign(
     service = _get_service()
     try:
         service.deactivate_campaign(campaign_id, owner_id)
-        return {"success": True, "message": f"Campaign '{campaign_id}' deactivated", "application_id": _app_id()}
+        return {"success": True, "message": f"Campaign '{campaign_id}' deactivated"}
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
     except AuthorizationError as e:
@@ -545,7 +527,7 @@ def deactivate_link(
     service = _get_service()
     try:
         service.deactivate_link(short_code, owner_id)
-        return {"success": True, "message": f"Link '{short_code}' deactivated", "application_id": _app_id()}
+        return {"success": True, "message": f"Link '{short_code}' deactivated"}
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
     except AuthorizationError as e:
@@ -580,7 +562,6 @@ def get_link_stats(
             custom_domain=stats.get("custom_domain"),
             fallback_url=stats.get("fallback_url"),
             webhook_url=stats.get("webhook_url"),
-            application_id=_app_id(),
         )
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
@@ -601,7 +582,7 @@ def get_click_details(
     service = _get_service()
     try:
         clicks = service.get_click_details(short_code, owner_id, limit)
-        return ClickDetailsResponse(clicks=[ClickDetail(**c) for c in clicks], application_id=_app_id())
+        return ClickDetailsResponse(clicks=[ClickDetail(**c) for c in clicks], )
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
     except AuthorizationError as e:
@@ -619,7 +600,7 @@ def get_qr_code(
     service = _get_service()
     try:
         data = service.get_qr_code(short_code)
-        return QRCodeResponse(**data, application_id=_app_id())
+        return QRCodeResponse(**data, )
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
 
@@ -636,7 +617,7 @@ def link_health(
     service = _get_service()
     try:
         result = service.check_link_health(short_code, owner_id)
-        return HealthCheckResponse(**result, application_id=_app_id())
+        return HealthCheckResponse(**result, )
     except LinkNotFoundError as e:
         raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
     except AuthorizationError as e:

app/api/v1/vector_stores.py

@@ -64,7 +64,7 @@ async def create_vector_store(
     token: str = Depends(require_auth),
     vector_store_service: VectorStoreService = Depends(get_vector_store_service),
 ) -> VectorStoreResponse:
-    store_id, app_id = await vector_store_service.create_store(
+    store_id = await vector_store_service.create_store(
         name=body.name,
         description=body.description or "",
         metadata=body.metadata,
@@ -73,7 +73,6 @@ async def create_vector_store(
     return VectorStoreResponse(
         success=True,
         vector_store_id=store_id,
-        app_id=app_id,
         name=body.name,
         description=body.description,
         embedding_dimension=stats["embedding_dimension"],
@@ -102,7 +101,6 @@ async def list_vector_stores(
         stores.append(VectorStoreResponse(
             success=True,
             vector_store_id=r.store_id,
-            app_id=stats.get("app_id", r.store_id),
             name=r.name,
             description=r.description,
             embedding_dimension=stats.get("embedding_dimension", 0),
@@ -130,7 +128,6 @@ async def get_vector_store(
     return VectorStoreResponse(
         success=True,
         vector_store_id=stats["store_id"],
-        app_id=stats["app_id"],
         name=stats["name"],
         description=stats["description"],
         embedding_dimension=stats["embedding_dimension"],
@@ -159,7 +156,6 @@ async def delete_vector_store(
     return VectorStoreResponse(
         success=True,
         vector_store_id=store_id,
-        app_id="",
         name=record.name,
         document_count=0,
         embedding_dimension=0,

app/config.py

@@ -8,7 +8,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
-        env_file=".env",
+        env_file=[".env", ".env.local"],
         env_file_encoding="utf-8",
         extra="ignore",
     )
@@ -73,19 +73,16 @@ class Settings(BaseSettings):
     jwt_default_expiry_minutes: int = 30
     jwt_issuer: str = "all-api-collection"
 
-    tidb_instances: str = ""
-    tidb_ssl: bool = True
+    supabase_url: str = "http://localhost:8000"
+    supabase_anon_key: str = ""
+    supabase_service_role_key: str = ""
+    supabase_schema: str = "public"
     access_token_expire_minutes: int = 15
     refresh_token_expire_days: int = 7
     max_login_attempts: int = 5
     lockout_minutes: int = 15
-    application_id: str = ""
     admin_password: str = ""
 
-    @property
-    def tidb_instance_list(self) -> list[str]:
-        return [u.strip() for u in self.tidb_instances.split(",") if u.strip()]
-
     @property
     def max_upload_mb(self) -> int:
         return self.max_upload_bytes // (1024 * 1024)

app/core/auth/deps.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import logging
 from datetime import datetime
-from typing import Annotated, Optional
+from typing import Annotated, List, Optional
 
 import jwt
 from argon2 import PasswordHasher
@@ -11,124 +11,28 @@ from fastapi import Depends, HTTPException, Request, status
 from app.config import get_settings
 from app.core.auth.models import RefreshSession, User, _uuid
 from app.core.auth.models import _utcnow as _now
-from app.core.tidb_manager import TiDBManager, get_tidb_manager, set_tidb_manager
+from app.services.supabase import AuthRepository, SupabaseClient, get_supabase_client
+from app.services.supabase.repositories import _row_to_user
+from app.services.supabase.client import create_supabase_client
 
 logger = logging.getLogger("auth")
 
 _settings = get_settings()
 ph = PasswordHasher()
 
-AUTH_TABLES: dict[str, str] = {
-    "users": """
-        CREATE TABLE IF NOT EXISTS users (
-            id VARCHAR(36) PRIMARY KEY,
-            email VARCHAR(255) NOT NULL,
-            username VARCHAR(50) NULL,
-            full_name VARCHAR(255) NULL,
-            password_hash TEXT NOT NULL,
-            is_active TINYINT(1) NOT NULL DEFAULT 1,
-            is_verified TINYINT(1) NOT NULL DEFAULT 0,
-            failed_login_attempts INT NOT NULL DEFAULT 0,
-            locked_until DATETIME NULL,
-            last_login DATETIME NULL,
-            password_changed_at DATETIME NULL,
-            created_at DATETIME NOT NULL,
-            updated_at DATETIME NOT NULL,
-            deleted_at DATETIME NULL,
-            UNIQUE KEY uk_users_email (email),
-            UNIQUE KEY uk_users_username (username),
-            INDEX idx_users_email (email),
-            INDEX idx_users_username (username)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "roles": """
-        CREATE TABLE IF NOT EXISTS roles (
-            id VARCHAR(36) PRIMARY KEY,
-            name VARCHAR(50) NOT NULL,
-            description VARCHAR(255) NULL,
-            UNIQUE KEY uk_roles_name (name)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "permissions": """
-        CREATE TABLE IF NOT EXISTS permissions (
-            id VARCHAR(36) PRIMARY KEY,
-            code VARCHAR(100) NOT NULL,
-            description VARCHAR(255) NULL,
-            UNIQUE KEY uk_permissions_code (code),
-            INDEX idx_permissions_code (code)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "user_roles": """
-        CREATE TABLE IF NOT EXISTS user_roles (
-            user_id VARCHAR(36) NOT NULL,
-            role_id VARCHAR(36) NOT NULL,
-            PRIMARY KEY (user_id, role_id),
-            CONSTRAINT fk_ur_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
-            CONSTRAINT fk_ur_role FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "role_permissions": """
-        CREATE TABLE IF NOT EXISTS role_permissions (
-            role_id VARCHAR(36) NOT NULL,
-            permission_id VARCHAR(36) NOT NULL,
-            PRIMARY KEY (role_id, permission_id),
-            CONSTRAINT fk_rp_role FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE,
-            CONSTRAINT fk_rp_perm FOREIGN KEY (permission_id) REFERENCES permissions(id) ON DELETE CASCADE
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "refresh_sessions": """
-        CREATE TABLE IF NOT EXISTS refresh_sessions (
-            id VARCHAR(36) PRIMARY KEY,
-            user_id VARCHAR(36) NOT NULL,
-            token_key VARCHAR(64) NOT NULL,
-            token_hash VARCHAR(255) NOT NULL,
-            device_info VARCHAR(255) NULL,
-            ip_address VARCHAR(45) NULL,
-            expires_at DATETIME NOT NULL,
-            revoked_at DATETIME NULL,
-            created_at DATETIME NOT NULL,
-            updated_at DATETIME NOT NULL,
-            UNIQUE KEY uk_rs_token_key (token_key),
-            INDEX idx_rs_user_id (user_id),
-            INDEX idx_rs_token_key (token_key),
-            CONSTRAINT fk_rs_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-}
-
-
-def _create_manager() -> Optional[TiDBManager]:
-    urls = _settings.tidb_instance_list
-    if not urls:
-        logger.warning("No TiDB instances configured (TIDB_INSTANCES is empty)")
-        return None
-    import ssl
-    if _settings.tidb_ssl:
-        ctx = ssl.create_default_context()
-        ctx.check_hostname = False
-        ctx.verify_mode = ssl.CERT_NONE
-    else:
-        ctx = None
-    mgr = TiDBManager(urls, ssl=ctx)
-    return mgr
-
 
 async def init_auth_db():
-    mgr = _create_manager()
-    if mgr is None:
-        logger.error("Cannot initialize auth DB: no TiDB instances configured")
+    client = await create_supabase_client()
+    if client is None:
+        logger.error("Cannot initialize auth DB: Supabase not configured")
         return
-    await mgr.initialize_pools()
-    set_tidb_manager(mgr)
+    logger.info("Supabase auth client initialized")
 
-    await mgr.ensure_tables(AUTH_TABLES)
+    repo = AuthRepository(client)
 
-    roles_result = await mgr.fetchone("SELECT id FROM roles WHERE name = 'SuperAdmin'")
-    if not roles_result:
-        sa_role_id = _uuid()
-        admin_role_id = _uuid()
-        user_role_id = _uuid()
-        perm_ids = {}
+    sa_role = await repo.get_role_by_name("SuperAdmin")
+    if not sa_role:
+        perm_ids: List[str] = []
         for code, desc in [
             ("users:read", "Read users"),
             ("users:write", "Modify users"),
@@ -136,76 +40,63 @@ async def init_auth_db():
             ("admin:access", "Access admin panel"),
         ]:
             pid = _uuid()
-            await mgr.execute(
-                "INSERT INTO permissions (id, code, description) VALUES (%s, %s, %s)",
-                (pid, code, desc),
-            )
-            perm_ids[code] = pid
+            await client.insert("permissions", {
+                "id": pid,
+                "code": code,
+                "description": desc,
+            })
+            perm_ids.append(pid)
 
-        await mgr.execute(
-            "INSERT INTO roles (id, name, description) VALUES (%s, %s, %s)",
-            (sa_role_id, "SuperAdmin", "Full system access"),
-        )
-        await mgr.execute(
-            "INSERT INTO roles (id, name, description) VALUES (%s, %s, %s)",
-            (admin_role_id, "Admin", "Administrative access"),
-        )
-        await mgr.execute(
-            "INSERT INTO roles (id, name, description) VALUES (%s, %s, %s)",
-            (user_role_id, "User", "Standard user access"),
-        )
+        sa_role_id = _uuid()
+        admin_role_id = _uuid()
+        user_role_id = _uuid()
 
-        perms_list = list(perm_ids.values())
-        for pid in perms_list:
-            await mgr.execute(
-                "INSERT INTO role_permissions (role_id, permission_id) VALUES (%s, %s)",
-                (sa_role_id, pid),
-            )
-        for pid in perms_list[:-1]:
-            await mgr.execute(
-                "INSERT INTO role_permissions (role_id, permission_id) VALUES (%s, %s)",
-                (admin_role_id, pid),
-            )
-        await mgr.execute(
-            "INSERT INTO role_permissions (role_id, permission_id) VALUES (%s, %s)",
-            (user_role_id, perms_list[0]),
-        )
+        await client.insert("roles", {"id": sa_role_id, "name": "SuperAdmin", "description": "Full system access"})
+        await client.insert("roles", {"id": admin_role_id, "name": "Admin", "description": "Administrative access"})
+        await client.insert("roles", {"id": user_role_id, "name": "User", "description": "Standard user access"})
+
+        for pid in perm_ids:
+            await client.insert("role_permissions", {"role_id": sa_role_id, "permission_id": pid})
+        for pid in perm_ids[:-1]:
+            await client.insert("role_permissions", {"role_id": admin_role_id, "permission_id": pid})
+        await client.insert("role_permissions", {"role_id": user_role_id, "permission_id": perm_ids[0]})
 
     admin_email = "admin@example.com"
-    admin_result = await mgr.fetchone("SELECT id FROM users WHERE email = %s", (admin_email,))
-    if not admin_result:
+    admin_user = await repo.find_user_by_email(admin_email)
+    if not admin_user:
         admin_password = _settings.admin_password or "Admin123!"
         admin_id = _uuid()
         now = _now()
-        await mgr.execute(
-            "INSERT INTO users (id, email, full_name, password_hash, is_verified, created_at, updated_at) "
-            "VALUES (%s, %s, %s, %s, 1, %s, %s)",
-            (admin_id, admin_email, "System Administrator", ph.hash(admin_password), now, now),
-        )
-        role_result = await mgr.fetchone("SELECT id FROM roles WHERE name = 'SuperAdmin'")
-        if role_result:
-            await mgr.execute(
-                "INSERT INTO user_roles (user_id, role_id) VALUES (%s, %s)",
-                (admin_id, role_result["id"]),
-            )
+        await client.insert("users", {
+            "id": admin_id,
+            "email": admin_email,
+            "full_name": "System Administrator",
+            "password_hash": ph.hash(admin_password),
+            "is_verified": True,
+            "created_at": now.isoformat(),
+            "updated_at": now.isoformat(),
+        })
+        sa_role = await repo.get_role_by_name("SuperAdmin")
+        if sa_role:
+            await client.insert("user_roles", {"user_id": admin_id, "role_id": sa_role["id"]})
 
     from app.services.auth_service import AuthService
-    await AuthService.cleanup_expired_sessions(mgr)
+    await AuthService.cleanup_expired_sessions()
 
 
-async def get_db() -> TiDBManager:
-    mgr = get_tidb_manager()
-    if mgr is None:
+async def get_db() -> SupabaseClient:
+    client = get_supabase_client()
+    if client is None:
         raise HTTPException(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail="Database not initialized. Configure TIDB_INSTANCES.",
+            detail="Database not initialized. Configure SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY.",
         )
-    return mgr
+    return client
 
 
 async def get_current_user(
     request: Request,
-    db: Annotated[TiDBManager, Depends(get_db)],
+    db: Annotated[SupabaseClient, Depends(get_db)],
 ) -> User:
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
@@ -228,51 +119,22 @@ async def get_current_user(
     except jwt.PyJWTError:
         raise credentials_exception
 
-    row = await db.fetchone_all(
-        "SELECT * FROM users WHERE id = %s AND deleted_at IS NULL",
-        (user_id,),
-    )
-    if row is None:
-        raise credentials_exception
-    user = _row_to_user(row)
-    if not user.is_active:
+    repo = AuthRepository(db)
+    user = await repo.find_user_by_id(user_id)
+    if user is None or not user.is_active:
         raise credentials_exception
     return user
 
 
-def require_application_id(request: Request):
-    app_id = request.headers.get("X-Application-Id")
-    expected = _settings.application_id
-    if not expected:
-        return True
-    if not app_id:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Missing X-Application-Id header",
-        )
-    if app_id != expected:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Invalid application_id",
-        )
-    return True
-
-
 def require_permissions(*required_perms: str):
     async def permission_checker(
         user: Annotated[User, Depends(get_current_user)],
     ) -> User:
-        user_perms = set()
-        mgr = get_tidb_manager()
-        if mgr:
-            rows = await mgr.fetchall_all(
-                "SELECT p.code FROM permissions p "
-                "JOIN role_permissions rp ON rp.permission_id = p.id "
-                "JOIN user_roles ur ON ur.role_id = rp.role_id "
-                "WHERE ur.user_id = %s",
-                (user.id,),
-            )
-            user_perms = {r["code"] for r in rows}
+        client = get_supabase_client()
+        user_perms: set[str] = set()
+        if client:
+            repo = AuthRepository(client)
+            user_perms = set(await repo.get_user_permission_codes(user.id))
         missing = [p for p in required_perms if p not in user_perms]
         if missing:
             raise HTTPException(
@@ -282,53 +144,3 @@ def require_permissions(*required_perms: str):
         return user
 
     return permission_checker
-
-
-def _row_to_user(row: dict) -> User:
-    return User(
-        id=row["id"],
-        email=row["email"],
-        username=row.get("username"),
-        full_name=row.get("full_name"),
-        password_hash=row["password_hash"],
-        is_active=bool(row["is_active"]),
-        is_verified=bool(row["is_verified"]),
-        failed_login_attempts=row["failed_login_attempts"],
-        locked_until=_deserialize_dt(row.get("locked_until")),
-        last_login=_deserialize_dt(row.get("last_login")),
-        password_changed_at=_deserialize_dt(row.get("password_changed_at")),
-        created_at=_deserialize_dt(row["created_at"]),
-        updated_at=_deserialize_dt(row["updated_at"]),
-        deleted_at=_deserialize_dt(row.get("deleted_at")),
-    )
-
-
-def _row_to_refresh_session(row: dict) -> RefreshSession:
-    return RefreshSession(
-        id=row["id"],
-        user_id=row["user_id"],
-        token_key=row["token_key"],
-        token_hash=row["token_hash"],
-        device_info=row.get("device_info"),
-        ip_address=row.get("ip_address"),
-        expires_at=_deserialize_dt(row["expires_at"]),
-        revoked_at=_deserialize_dt(row.get("revoked_at")),
-        created_at=_deserialize_dt(row["created_at"]),
-        updated_at=_deserialize_dt(row["updated_at"]),
-    )
-
-
-def _serialize_dt(dt: Optional[datetime]) -> Optional[str]:
-    if dt is None:
-        return None
-    return dt.strftime("%Y-%m-%d %H:%M:%S.%f")
-
-
-def _deserialize_dt(val) -> Optional[datetime]:
-    if val is None:
-        return None
-    if isinstance(val, datetime):
-        return val
-    if isinstance(val, str):
-        return datetime.fromisoformat(val)
-    return val

app/core/auth/schemas.py

@@ -123,29 +123,23 @@ class ApiResponse(BaseModel):
 
 class DataResponse(ApiResponse):
     data: dict
-    application_id: Optional[str] = None
 
 
 class TokenResponse(ApiResponse):
     data: TokenData
-    application_id: Optional[str] = None
 
 
 class ProfileResponse(ApiResponse):
     data: UserProfile
-    application_id: Optional[str] = None
 
 
 class MessageDataResponse(ApiResponse):
     data: MessageResponse
-    application_id: Optional[str] = None
 
 
 class SchemaResponse(ApiResponse):
     data: UserSchemaResponse
-    application_id: Optional[str] = None
 
 
 class SessionListResponse(ApiResponse):
     data: list[SessionOut]
-    application_id: Optional[str] = None

app/core/banner.py

@@ -13,10 +13,10 @@ def get_banner() -> str:
         f"  {settings.app_name} v{settings.app_version}",
         f"  Environment: {settings.environment or 'production'}",
     ]
-    if settings.tidb_instance_list:
-        lines.append(f"  TiDB instances: {len(settings.tidb_instance_list)}")
+    if settings.supabase_url and settings.supabase_service_role_key:
+        lines.append("  Supabase: connected")
     else:
-        lines.append("  TiDB: not configured")
+        lines.append("  Supabase: not configured")
     if settings.redis_url:
         lines.append("  Redis: connected")
     else:

app/core/tidb_manager.py

@@ -1,341 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-import logging
-import ssl
-from typing import Any, Dict, List, Optional, Set
-from urllib.parse import urlparse
-
-import aiomysql
-
-logger = logging.getLogger(__name__)
-
-# MySQL/TiDB error codes that indicate the instance cannot accept more writes:
-#   1114  (ER_RECORD_FILE_FULL)  – "The table '...' is full"
-#   1021  (ER_DISK_FULL)         – "Disk full (...)"
-#   126   (ER_INDEX_FILE_FULL)   – "Index file is full"
-#   1877  (ER_TABLE_FULL)        – "Table full"
-_STORAGE_FULL_CODES: Set[int] = {1114, 1021, 126, 1877}
-
-
-class TiDBWriteError(RuntimeError):
-    """All TiDB instances are full or unavailable — writes cannot be accepted."""
-    def __init__(self):
-        super().__init__("No TiDB instances available for write. All instances are full or unreachable.")
-
-
-def _is_storage_full_error(exc: Exception) -> bool:
-    if isinstance(exc, aiomysql.MySQLError):
-        code = exc.args[0] if exc.args else None
-        if isinstance(code, int) and code in _STORAGE_FULL_CODES:
-            return True
-        msg = str(exc).lower()
-        for keyword in ("table is full", "disk full", "storage full",
-                        "quota exceeded", "out of storage"):
-            if keyword in msg:
-                return True
-    return False
-
-
-class TiDBInstance:
-    def __init__(self, host: str, port: int, user: str, password: str, database: str,
-                 ssl: Optional[ssl.SSLContext] = None):
-        self.host = host
-        self.port = port
-        self.user = user
-        self.password = password
-        self.database = database
-        self.ssl = ssl
-        self.pool: Optional[aiomysql.Pool] = None
-        self.is_full = False
-        self.is_available = True
-
-    @classmethod
-    def from_url(cls, url: str,
-                 ssl: Optional[ssl.SSLContext] = None) -> TiDBInstance:
-        parsed = urlparse(url)
-        host = parsed.hostname or "localhost"
-        port = parsed.port or 4000
-        user = parsed.username or "root"
-        password = parsed.password or ""
-        database = parsed.path.lstrip("/") or "test"
-        return cls(host=host, port=port, user=user, password=password,
-                   database=database, ssl=ssl)
-
-    async def create_pool(self, minsize: int = 1, maxsize: int = 10):
-        self.pool = await aiomysql.create_pool(
-            host=self.host,
-            port=self.port,
-            user=self.user,
-            password=self.password,
-            db=self.database,
-            minsize=minsize,
-            maxsize=maxsize,
-            autocommit=False,
-            cursorclass=aiomysql.DictCursor,
-            ssl=self.ssl,
-        )
-        self.is_available = True
-        return self.pool
-
-    async def close_pool(self):
-        if self.pool:
-            await self.pool.close()
-            await self.pool.wait_closed()
-            self.pool = None
-
-    def __repr__(self) -> str:
-        user_prefix = self.user.split(".")[0] if "." in self.user else self.user
-        return f"TiDBInstance({user_prefix}@...)"
-
-
-class TiDBManager:
-    def __init__(self, instance_urls: List[str],
-                 ssl: Optional[ssl.SSLContext] = None):
-        self.instances = [TiDBInstance.from_url(url, ssl=ssl) for url in instance_urls]
-        self._current_instance_index = 0
-
-    @property
-    def current_instance(self) -> TiDBInstance:
-        return self.instances[self._current_instance_index]
-
-    @property
-    def current_index(self) -> int:
-        return self._current_instance_index
-
-    async def initialize_pools(self, minsize: int = 1, maxsize: int = 10):
-        for i, inst in enumerate(self.instances, 1):
-            try:
-                await inst.create_pool(minsize=minsize, maxsize=maxsize)
-                logger.info("Connected to TiDB instance %d/%d", i, len(self.instances))
-            except Exception as exc:
-                logger.error("Failed to connect to TiDB instance %d/%d: %s", i, len(self.instances), exc)
-                inst.is_available = False
-
-    async def close_all_pools(self):
-        for inst in self.instances:
-            await inst.close_pool()
-
-    async def ensure_tables(self, table_ddls: Dict[str, str]):
-        for i, inst in enumerate(self.instances, 1):
-            if not inst.is_available or not inst.pool:
-                continue
-            async with inst.pool.acquire() as conn:
-                async with conn.cursor() as cur:
-                    for table_name, ddl in table_ddls.items():
-                        try:
-                            await cur.execute(
-                                "SELECT 1 FROM information_schema.tables "
-                                "WHERE table_schema = %s AND table_name = %s",
-                                (inst.database, table_name),
-                            )
-                            exists = await cur.fetchone()
-                            if not exists:
-                                for statement in ddl.split(";"):
-                                    stmt = statement.strip()
-                                    if stmt:
-                                        await cur.execute(stmt)
-                                logger.info("Created table '%s' (instance %d/%d)", table_name, i, len(self.instances))
-                        except Exception as exc:
-                            logger.error(
-                                "Failed to create table '%s' (instance %d/%d): %s",
-                                table_name, i, len(self.instances), exc,
-                            )
-                await conn.commit()
-
-    async def has_table(self, table_name: str) -> bool:
-        inst = self.current_instance
-        if not inst.pool:
-            return False
-        async with inst.pool.acquire() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(
-                    "SELECT 1 FROM information_schema.tables "
-                    "WHERE table_schema = %s AND table_name = %s",
-                    (inst.database, table_name),
-                )
-                return await cur.fetchone() is not None
-
-    async def check_and_rotate(self) -> bool:
-        if self.current_instance.is_available and not self.current_instance.is_full:
-            return True
-
-        for i in range(1, len(self.instances)):
-            idx = (self._current_instance_index + i) % len(self.instances)
-            inst = self.instances[idx]
-            if inst.is_available and not inst.is_full:
-                logger.info(
-                    "Rotating TiDB write target: instance %d -> %d",
-                    self._current_instance_index + 1, idx + 1,
-                )
-                self._current_instance_index = idx
-                return True
-
-        logger.error("All TiDB instances are full or unavailable")
-        return False
-
-    async def fetchone(self, query: str, params=None) -> Optional[Dict[str, Any]]:
-        inst = self.current_instance
-        if not inst.pool:
-            return None
-        async with inst.pool.acquire() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(query, params or ())
-                return await cur.fetchone()
-
-    async def fetchone_all(self, query: str, params=None) -> Optional[Dict[str, Any]]:
-        async def _query(inst_: TiDBInstance):
-            try:
-                async with inst_.pool.acquire() as conn:
-                    async with conn.cursor() as cur:
-                        await cur.execute(query, params or ())
-                        return await cur.fetchone()
-            except asyncio.CancelledError:
-                raise
-            except Exception as exc:
-                logger.error("Error reading from %s: %s", inst_, exc)
-                return None
-
-        tasks = [
-            asyncio.create_task(_query(inst))
-            for inst in self.instances
-            if inst.pool and inst.is_available
-        ]
-        if not tasks:
-            return None
-
-        for coro in asyncio.as_completed(tasks):
-            result = await coro
-            if result is not None:
-                for t in tasks:
-                    if not t.done():
-                        t.cancel()
-                return result
-        return None
-
-    async def fetchall(self, query: str, params=None) -> List[Dict[str, Any]]:
-        inst = self.current_instance
-        if not inst.pool:
-            return []
-        async with inst.pool.acquire() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(query, params or ())
-                return await cur.fetchall()
-
-    async def fetchall_all(self, query: str, params=None) -> List[Dict[str, Any]]:
-        async def _query(inst_: TiDBInstance):
-            try:
-                async with inst_.pool.acquire() as conn:
-                    async with conn.cursor() as cur:
-                        await cur.execute(query, params or ())
-                        return await cur.fetchall()
-            except asyncio.CancelledError:
-                raise
-            except Exception as exc:
-                logger.error("Error reading from %s: %s", inst_, exc)
-                return []
-
-        tasks = [
-            asyncio.create_task(_query(inst))
-            for inst in self.instances
-            if inst.pool and inst.is_available
-        ]
-        if not tasks:
-            return []
-
-        all_rows = await asyncio.gather(*tasks)
-        seen = set()
-        merged = []
-        for rows in all_rows:
-            for row in rows:
-                key = tuple(row.items())
-                if key not in seen:
-                    seen.add(key)
-                    merged.append(row)
-        return merged
-
-    async def _try_write(self, query: str, params) -> int:
-        inst = self.current_instance
-        if not inst.pool:
-            raise RuntimeError(f"TiDB pool not initialized for {inst}")
-        async with inst.pool.acquire() as conn:
-            async with conn.cursor(aiomysql.DictCursor) as cur:
-                await cur.execute(query, params or ())
-                await conn.commit()
-                return cur.lastrowid if cur.lastrowid is not None else 0
-
-    async def execute(self, query: str, params=None) -> int:
-        ok = await self.check_and_rotate()
-        if not ok:
-            raise TiDBWriteError()
-        try:
-            return await self._try_write(query, params)
-        except Exception as exc:
-            if _is_storage_full_error(exc):
-                logger.warning("TiDB write target is full, rotating...")
-                self.current_instance.is_full = True
-                ok = await self.check_and_rotate()
-                if not ok:
-                    raise TiDBWriteError() from exc
-                try:
-                    return await self._try_write(query, params)
-                except Exception:
-                    raise TiDBWriteError() from exc
-            raise
-
-    async def execute_on_all(self, query: str, params=None):
-        for i, inst in enumerate(self.instances, 1):
-            if not inst.is_available or not inst.pool:
-                continue
-            try:
-                async with inst.pool.acquire() as conn:
-                    async with conn.cursor(aiomysql.DictCursor) as cur:
-                        await cur.execute(query, params or ())
-                    await conn.commit()
-            except Exception as exc:
-                logger.error("Error executing on instance %d/%d: %s", i, len(self.instances), exc)
-
-    async def insert_and_get_id(self, query: str, params=None) -> int:
-        ok = await self.check_and_rotate()
-        if not ok:
-            raise TiDBWriteError()
-        try:
-            return await self._try_write(query, params)
-        except Exception as exc:
-            if _is_storage_full_error(exc):
-                logger.warning("TiDB write target is full, rotating...")
-                self.current_instance.is_full = True
-                ok = await self.check_and_rotate()
-                if not ok:
-                    raise TiDBWriteError() from exc
-                try:
-                    return await self._try_write(query, params)
-                except Exception:
-                    raise TiDBWriteError() from exc
-            raise
-
-    def get_instances_summary(self) -> List[Dict[str, Any]]:
-        return [
-            {
-                "host": inst.host,
-                "port": inst.port,
-                "database": inst.database,
-                "is_available": inst.is_available,
-                "is_full": inst.is_full,
-                "is_current": i == self._current_instance_index,
-            }
-            for i, inst in enumerate(self.instances)
-        ]
-
-
-_tidb_manager: Optional[TiDBManager] = None
-
-
-def get_tidb_manager() -> Optional[TiDBManager]:
-    global _tidb_manager
-    return _tidb_manager
-
-
-def set_tidb_manager(mgr: TiDBManager):
-    global _tidb_manager
-    _tidb_manager = mgr

app/core/vector_store/__init__.py

@@ -1,17 +1,8 @@
-from __future__ import annotations
-
-from app.core.vector_store.deps import (
-    get_vs_db,
-    get_vs_manager,
-    init_vector_store_db,
-    set_vs_manager,
-)
+from app.core.vector_store.deps import get_vs_db, init_vector_store_db
 from app.core.vector_store.models import VectorStoreIndex
 
 __all__ = [
     "VectorStoreIndex",
     "init_vector_store_db",
     "get_vs_db",
-    "set_vs_manager",
-    "get_vs_manager",
 ]

app/core/vector_store/deps.py

@@ -4,49 +4,21 @@ import logging
 from typing import Optional
 
 from app.config import get_settings
-from app.core.tidb_manager import TiDBManager, get_tidb_manager
+from app.services.supabase import SupabaseClient, get_supabase_client
 
 logger = logging.getLogger(__name__)
 
-_settings = get_settings()
-
-VS_TABLES: dict[str, str] = {
-    "vector_store_index": """
-        CREATE TABLE IF NOT EXISTS vector_store_index (
-            store_id VARCHAR(36) PRIMARY KEY,
-            name VARCHAR(255) NOT NULL,
-            path VARCHAR(1024) NOT NULL,
-            description VARCHAR(1024) DEFAULT '',
-            metadata_json TEXT,
-            created_at VARCHAR(64) NOT NULL
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-}
-
 
 async def init_vector_store_db():
-    mgr = get_tidb_manager()
-    if mgr is None:
-        logger.error("Cannot init vector store DB: TiDB manager not initialized")
+    client = get_supabase_client()
+    if client is None:
+        logger.error("Cannot init vector store DB: Supabase not initialized")
         return
-    await mgr.ensure_tables(VS_TABLES)
-    logger.info("Vector store tables ensured on all TiDB instances")
-
-
-async def get_vs_db() -> TiDBManager:
-    mgr = get_tidb_manager()
-    if mgr is None:
-        raise RuntimeError("TiDB manager not initialized")
-    return mgr
-
-
-_tidb_vs_manager: Optional[TiDBManager] = None
-
-
-def set_vs_manager(mgr: TiDBManager):
-    global _tidb_vs_manager
-    _tidb_vs_manager = mgr
+    logger.info("Vector store DB initialized via Supabase")
 
 
-def get_vs_manager() -> Optional[TiDBManager]:
-    return _tidb_vs_manager
+async def get_vs_db() -> SupabaseClient:
+    client = get_supabase_client()
+    if client is None:
+        raise RuntimeError("Supabase client not initialized")
+    return client

app/models/schemas.py

@@ -568,7 +568,6 @@ class VectorStoreCreate(BaseModel):
 class VectorStoreResponse(BaseModel):
     success: bool
     vector_store_id: str
-    app_id: str
     name: str
     description: Optional[str] = None
     embedding_dimension: int

app/services/auth_service.py

@@ -11,7 +11,6 @@ from argon2.exceptions import VerifyMismatchError
 from fastapi import HTTPException, Request
 
 from app.config import get_settings
-from app.core.auth.deps import _row_to_refresh_session, _row_to_user, _serialize_dt
 from app.core.auth.models import RefreshSession, User, _uuid
 from app.core.auth.models import _utcnow as _now
 from app.core.auth.schemas import (
@@ -24,7 +23,7 @@ from app.core.auth.schemas import (
     UpdateProfileSchema,
     UserProfile,
 )
-from app.core.tidb_manager import TiDBManager
+from app.services.supabase import AuthRepository, SupabaseClient
 
 logger = logging.getLogger("auth_service")
 _settings = get_settings()
@@ -38,50 +37,44 @@ def _token_key(raw: str) -> str:
 class AuthService:
 
     @staticmethod
-    async def register(db: TiDBManager, schema: RegisterSchema) -> User:
-        existing = await db.fetchone_all(
-            "SELECT id FROM users WHERE email = %s AND deleted_at IS NULL",
-            (schema.email,),
-        )
+    async def register(db: SupabaseClient, schema: RegisterSchema) -> User:
+        repo = AuthRepository(db)
+
+        existing = await repo.find_user_by_email(schema.email)
         if existing:
             raise HTTPException(status_code=409, detail="Email already registered")
 
         if schema.username:
-            existing = await db.fetchone_all(
-                "SELECT id FROM users WHERE username = %s AND deleted_at IS NULL",
-                (schema.username,),
-            )
+            existing = await repo.find_user_by_username(schema.username)
             if existing:
                 raise HTTPException(status_code=409, detail="Username already taken")
 
         user_id = _uuid()
         now = _now()
-        await db.insert_and_get_id(
-            "INSERT INTO users (id, email, username, full_name, password_hash, created_at, updated_at) "
-            "VALUES (%s, %s, %s, %s, %s, %s, %s)",
-            (user_id, schema.email, schema.username, schema.full_name, ph.hash(schema.password), now, now),
+        user = User(
+            id=user_id,
+            email=schema.email,
+            username=schema.username,
+            full_name=schema.full_name,
+            password_hash=ph.hash(schema.password),
+            created_at=now,
+            updated_at=now,
         )
+        await repo.create_user(user)
 
-        role_row = await db.fetchone("SELECT id FROM roles WHERE name = 'User'")
+        role_row = await repo.get_role_by_name("User")
         if role_row:
-            await db.execute(
-                "INSERT INTO user_roles (user_id, role_id) VALUES (%s, %s)",
-                (user_id, role_row["id"]),
-            )
+            await repo.assign_role(user_id, role_row["id"])
 
-        user_row = await db.fetchone("SELECT * FROM users WHERE id = %s", (user_id,))
-        return _row_to_user(user_row)
+        return await repo.find_user_by_id(user_id)
 
     @staticmethod
-    async def login(db: TiDBManager, request: Request, schema: LoginSchema) -> TokenData:
-        row = await db.fetchone_all(
-            "SELECT * FROM users WHERE email = %s AND deleted_at IS NULL",
-            (schema.email,),
-        )
-        if not row:
-            raise HTTPException(status_code=401, detail="Incorrect email or password")
+    async def login(db: SupabaseClient, request: Request, schema: LoginSchema) -> TokenData:
+        repo = AuthRepository(db)
 
-        user = _row_to_user(row)
+        user = await repo.find_user_by_email(schema.email)
+        if not user:
+            raise HTTPException(status_code=401, detail="Incorrect email or password")
 
         if user.locked_until and user.locked_until > _now():
             raise HTTPException(
@@ -93,148 +86,118 @@ class AuthService:
             user.failed_login_attempts += 1
             if user.failed_login_attempts >= _settings.max_login_attempts:
                 user.locked_until = _now() + timedelta(minutes=_settings.lockout_minutes)
-            await db.execute(
-                "UPDATE users SET failed_login_attempts = %s, locked_until = %s WHERE id = %s",
-                (user.failed_login_attempts, _serialize_dt(user.locked_until), user.id),
-            )
+            await repo.update_user(user.id, {
+                "failed_login_attempts": user.failed_login_attempts,
+                "locked_until": user.locked_until,
+            })
             raise HTTPException(status_code=401, detail="Incorrect email or password")
 
         now = _now()
-        await db.execute(
-            "UPDATE users SET failed_login_attempts = 0, locked_until = NULL, last_login = %s "
-            "WHERE id = %s",
-            (now, user.id),
-        )
+        await repo.update_user(user.id, {
+            "failed_login_attempts": 0,
+            "locked_until": None,
+            "last_login": now,
+        })
 
         access_token = AuthService._create_access_token(user.id)
         raw_refresh, refresh_hash, token_key, expires_at = AuthService._create_refresh_token()
 
         session_id = _uuid()
-        await db.execute(
-            "INSERT INTO refresh_sessions (id, user_id, token_key, token_hash, device_info, "
-            "ip_address, expires_at, created_at, updated_at) "
-            "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
-            (
-                session_id, user.id, token_key, refresh_hash,
-                request.headers.get("User-Agent", "Unknown"),
-                request.client.host if request.client else "Unknown",
-                _serialize_dt(expires_at), now, now,
-            ),
+        session = RefreshSession(
+            id=session_id,
+            user_id=user.id,
+            token_key=token_key,
+            token_hash=refresh_hash,
+            device_info=request.headers.get("User-Agent", "Unknown"),
+            ip_address=request.client.host if request.client else "Unknown",
+            expires_at=expires_at,
+            created_at=now,
+            updated_at=now,
         )
+        await repo.create_refresh_session(session)
 
         return TokenData(access_token=access_token, refresh_token=raw_refresh)
 
     @staticmethod
-    async def refresh(db: TiDBManager, raw_refresh_token: str) -> TokenData:
+    async def refresh(db: SupabaseClient, raw_refresh_token: str) -> TokenData:
+        repo = AuthRepository(db)
         key = _token_key(raw_refresh_token)
 
-        row = await db.fetchone_all(
-            "SELECT * FROM refresh_sessions WHERE token_key = %s AND revoked_at IS NULL "
-            "AND expires_at > %s",
-            (key, _serialize_dt(_now())),
-        )
-        if not row:
-            comp = await db.fetchone_all(
-                "SELECT * FROM refresh_sessions WHERE token_key = %s", (key,),
-            )
-            if comp and comp.get("revoked_at") is not None:
-                now = _serialize_dt(_now())
-                await db.execute_on_all(
-                    "UPDATE refresh_sessions SET revoked_at = %s "
-                    "WHERE user_id = %s AND revoked_at IS NULL",
-                    (now, comp["user_id"]),
-                )
+        session = await repo.find_valid_session(key)
+        if not session:
+            comp = await repo.find_session_by_token_key(key)
+            if comp and comp.revoked_at is not None:
+                await repo.revoke_all_user_sessions(comp.user_id)
                 raise HTTPException(
                     status_code=401,
                     detail="Session compromised. All sessions revoked. Please login again.",
                 )
             raise HTTPException(status_code=401, detail="Invalid or expired refresh token")
 
-        session = _row_to_refresh_session(row)
-
         if not AuthService._verify_token(raw_refresh_token, session.token_hash):
             raise HTTPException(status_code=401, detail="Invalid refresh token")
 
-        user_row = await db.fetchone_all(
-            "SELECT * FROM users WHERE id = %s AND deleted_at IS NULL AND is_active = 1",
-            (session.user_id,),
-        )
-        if not user_row:
+        user = await repo.find_user_by_id(session.user_id)
+        if not user or not user.is_active:
             raise HTTPException(status_code=401, detail="User not found or inactive")
-        user = _row_to_user(user_row)
 
-        now = _serialize_dt(_now())
-        await db.execute(
-            "UPDATE refresh_sessions SET revoked_at = %s WHERE id = %s",
-            (now, session.id),
-        )
+        now = _now()
+        await repo.revoke_session(session.id)
 
         new_access = AuthService._create_access_token(user.id)
         new_raw_refresh, new_hash, new_key, new_expires = AuthService._create_refresh_token()
 
         new_session_id = _uuid()
-        await db.execute(
-            "INSERT INTO refresh_sessions (id, user_id, token_key, token_hash, device_info, "
-            "ip_address, expires_at, created_at, updated_at) "
-            "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
-            (
-                new_session_id, user.id, new_key, new_hash,
-                session.device_info, session.ip_address,
-                _serialize_dt(new_expires), _serialize_dt(_now()), _serialize_dt(_now()),
-            ),
+        new_session = RefreshSession(
+            id=new_session_id,
+            user_id=user.id,
+            token_key=new_key,
+            token_hash=new_hash,
+            device_info=session.device_info,
+            ip_address=session.ip_address,
+            expires_at=new_expires,
+            created_at=now,
+            updated_at=now,
         )
+        await repo.create_refresh_session(new_session)
 
-        await db.execute(
-            "UPDATE users SET last_login = %s WHERE id = %s",
-            (_serialize_dt(_now()), user.id),
-        )
+        await repo.update_user(user.id, {"last_login": now})
 
         return TokenData(access_token=new_access, refresh_token=new_raw_refresh)
 
     @staticmethod
-    async def logout(db: TiDBManager, user: User, raw_refresh_token: str):
+    async def logout(db: SupabaseClient, user: User, raw_refresh_token: str):
+        repo = AuthRepository(db)
         key = _token_key(raw_refresh_token)
-        row = await db.fetchone_all(
-            "SELECT id FROM refresh_sessions WHERE token_key = %s "
-            "AND user_id = %s AND revoked_at IS NULL",
-            (key, user.id),
-        )
-        if not row:
+        session = await repo.find_session_by_token_key(key)
+        if not session or session.user_id != user.id:
             raise HTTPException(status_code=404, detail="Session not found")
-        await db.execute(
-            "UPDATE refresh_sessions SET revoked_at = %s WHERE id = %s",
-            (_serialize_dt(_now()), row["id"]),
-        )
+        await repo.revoke_session(session.id)
 
     @staticmethod
-    async def logout_all(db: TiDBManager, user: User):
-        now = _serialize_dt(_now())
-        await db.execute(
-            "UPDATE refresh_sessions SET revoked_at = %s "
-            "WHERE user_id = %s AND revoked_at IS NULL",
-            (now, user.id),
-        )
+    async def logout_all(db: SupabaseClient, user: User):
+        repo = AuthRepository(db)
+        await repo.revoke_all_user_sessions(user.id)
 
     @staticmethod
-    async def change_password(db: TiDBManager, user: User, schema: ChangePasswordSchema):
+    async def change_password(db: SupabaseClient, user: User, schema: ChangePasswordSchema):
         if not AuthService._verify_password(schema.current_password, user.password_hash):
             raise HTTPException(status_code=400, detail="Incorrect current password")
-        now = _serialize_dt(_now())
-        await db.execute(
-            "UPDATE users SET password_hash = %s, password_changed_at = %s WHERE id = %s",
-            (ph.hash(schema.new_password), now, user.id),
-        )
+        repo = AuthRepository(db)
+        now = _now()
+        await repo.update_user(user.id, {
+            "password_hash": ph.hash(schema.new_password),
+            "password_changed_at": now,
+        })
 
     @staticmethod
-    async def forgot_password(db: TiDBManager, schema: ForgotPasswordSchema):
-        row = await db.fetchone_all(
-            "SELECT id FROM users WHERE email = %s AND deleted_at IS NULL",
-            (schema.email,),
-        )
-        if row:
+    async def forgot_password(db: SupabaseClient, schema: ForgotPasswordSchema):
+        repo = AuthRepository(db)
+        user = await repo.find_user_by_email(schema.email)
+        if user:
             reset_token = jwt.encode(
                 {
-                    "sub": row["id"],
+                    "sub": user.id,
                     "type": "reset_password",
                     "exp": _now() + timedelta(hours=1),
                 },
@@ -244,7 +207,7 @@ class AuthService:
             logger.info("Password reset token for %s: %s", schema.email, reset_token)
 
     @staticmethod
-    async def reset_password(db: TiDBManager, schema: ResetPasswordSchema):
+    async def reset_password(db: SupabaseClient, schema: ResetPasswordSchema):
         try:
             payload = jwt.decode(
                 schema.token,
@@ -257,90 +220,63 @@ class AuthService:
         except jwt.PyJWTError:
             raise HTTPException(status_code=400, detail="Invalid or expired reset token")
 
-        row = await db.fetchone_all("SELECT id FROM users WHERE id = %s", (user_id,))
-        if not row:
+        repo = AuthRepository(db)
+        user = await repo.find_user_by_id(user_id)
+        if not user:
             raise HTTPException(status_code=404, detail="User not found")
 
-        now = _serialize_dt(_now())
-        await db.execute(
-            "UPDATE users SET password_hash = %s, password_changed_at = %s WHERE id = %s",
-            (ph.hash(schema.new_password), now, user_id),
-        )
-        await db.execute(
-            "UPDATE refresh_sessions SET revoked_at = %s "
-            "WHERE user_id = %s AND revoked_at IS NULL",
-            (now, user_id),
-        )
+        now = _now()
+        await repo.update_user(user_id, {
+            "password_hash": ph.hash(schema.new_password),
+            "password_changed_at": now,
+        })
+        await repo.revoke_all_user_sessions(user_id)
 
     @staticmethod
-    async def update_profile(db: TiDBManager, user: User, schema: UpdateProfileSchema) -> User:
+    async def update_profile(db: SupabaseClient, user: User, schema: UpdateProfileSchema) -> User:
+        repo = AuthRepository(db)
         if schema.username is not None:
-            existing = await db.fetchone_all(
-                "SELECT id FROM users WHERE username = %s AND id != %s AND deleted_at IS NULL",
-                (schema.username, user.id),
-            )
-            if existing:
+            existing = await repo.find_user_by_username(schema.username)
+            if existing and existing.id != user.id:
                 raise HTTPException(status_code=409, detail="Username already taken")
-            await db.execute("UPDATE users SET username = %s WHERE id = %s", (schema.username, user.id))
+            await repo.update_user(user.id, {"username": schema.username})
         if schema.full_name is not None:
-            await db.execute(
-                "UPDATE users SET full_name = %s WHERE id = %s",
-                (schema.full_name, user.id),
-            )
-        row = await db.fetchone("SELECT * FROM users WHERE id = %s", (user.id,))
-        return _row_to_user(row)
+            await repo.update_user(user.id, {"full_name": schema.full_name})
+        return await repo.find_user_by_id(user.id)
 
     @staticmethod
-    async def soft_delete(db: TiDBManager, user: User):
-        now = _serialize_dt(_now())
-        await db.execute(
-            "UPDATE users SET deleted_at = %s, is_active = 0 WHERE id = %s",
-            (now, user.id),
-        )
-        await db.execute(
-            "UPDATE refresh_sessions SET revoked_at = %s "
-            "WHERE user_id = %s AND revoked_at IS NULL",
-            (now, user.id),
-        )
+    async def soft_delete(db: SupabaseClient, user: User):
+        repo = AuthRepository(db)
+        now = _now()
+        await repo.update_user(user.id, {
+            "deleted_at": now,
+            "is_active": False,
+        })
+        await repo.revoke_all_user_sessions(user.id)
 
     @staticmethod
-    async def list_sessions(db: TiDBManager, user: User) -> list[RefreshSession]:
-        rows = await db.fetchall_all(
-            "SELECT * FROM refresh_sessions WHERE user_id = %s AND revoked_at IS NULL "
-            "ORDER BY created_at DESC",
-            (user.id,),
-        )
-        return [_row_to_refresh_session(r) for r in rows]
+    async def list_sessions(db: SupabaseClient, user: User) -> list[RefreshSession]:
+        repo = AuthRepository(db)
+        return await repo.find_user_sessions(user.id)
 
     @staticmethod
-    async def revoke_session(db: TiDBManager, user: User, session_id: str):
-        row = await db.fetchone_all(
-            "SELECT id FROM refresh_sessions WHERE id = %s AND user_id = %s AND revoked_at IS NULL",
-            (session_id, user.id),
-        )
-        if not row:
+    async def revoke_session(db: SupabaseClient, user: User, session_id: str):
+        repo = AuthRepository(db)
+        sessions = await repo.find_user_sessions(user.id)
+        target = [s for s in sessions if s.id == session_id]
+        if not target:
             raise HTTPException(status_code=404, detail="Session not found")
-        await db.execute(
-            "UPDATE refresh_sessions SET revoked_at = %s WHERE id = %s",
-            (_serialize_dt(_now()), session_id),
-        )
+        await repo.revoke_session(session_id)
 
     @staticmethod
-    async def cleanup_expired_sessions(db: TiDBManager):
-        rows = await db.fetchall_all(
-            "SELECT * FROM refresh_sessions WHERE expires_at < %s",
-            (_serialize_dt(_now()),),
-        )
-        count = 0
-        for row in rows:
-            if row.get("revoked_at") is None:
-                await db.execute(
-                    "UPDATE refresh_sessions SET revoked_at = %s WHERE id = %s",
-                    (_serialize_dt(_now()), row["id"]),
-                )
-                count += 1
-        if count:
-            logger.info("Cleaned up %s expired sessions", count)
+    async def cleanup_expired_sessions():
+        from app.services.supabase import get_supabase_client
+        client = get_supabase_client()
+        if client:
+            repo = AuthRepository(client)
+            count = await repo.cleanup_expired_sessions()
+            if count:
+                logger.info("Cleaned up %s expired sessions", count)
 
     @staticmethod
     def _create_access_token(user_id: str) -> str:
@@ -379,17 +315,12 @@ class AuthService:
             return False
 
     @staticmethod
-    async def user_to_profile(db: TiDBManager, user: User) -> UserProfile:
-        roles = []
+    async def user_to_profile(db: SupabaseClient, user: User) -> UserProfile:
+        roles: list[str] = []
         if db:
             try:
-                rows = await db.fetchall_all(
-                    "SELECT r.name FROM roles r "
-                    "JOIN user_roles ur ON ur.role_id = r.id "
-                    "WHERE ur.user_id = %s",
-                    (user.id,),
-                )
-                roles = [r["name"] for r in rows]
+                repo = AuthRepository(db)
+                roles = await repo.get_user_roles(user.id)
             except Exception:
                 pass
         return UserProfile(

app/services/supabase/__init__.py

@@ -0,0 +1,13 @@
+from app.services.supabase.client import SupabaseClient, get_supabase_client, set_supabase_client
+from app.services.supabase.repositories import (
+    AuthRepository,
+    VectorStoreRepository,
+)
+
+__all__ = [
+    "SupabaseClient",
+    "get_supabase_client",
+    "set_supabase_client",
+    "AuthRepository",
+    "VectorStoreRepository",
+]

app/services/supabase/client.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Dict, List, Optional
+
+from supabase import create_client, Client
+
+from app.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+
+class SupabaseClient:
+    def __init__(self, url: str, service_key: str):
+        self._url = url
+        self._service_key = service_key
+        self._client: Optional[Client] = None
+
+    async def initialize(self):
+        self._client = await asyncio.to_thread(create_client, self._url, self._service_key)
+        logger.info("Supabase client initialized")
+
+    @property
+    def client(self) -> Client:
+        if self._client is None:
+            raise RuntimeError("Supabase client not initialized. Call initialize() first.")
+        return self._client
+
+    async def close(self):
+        self._client = None
+        set_supabase_client(None)
+        logger.info("Supabase client closed")
+
+    # --- Table operations ---
+
+    async def select(
+        self, table: str, columns: str = "*",
+        eq: Optional[tuple[str, Any]] = None,
+        order: Optional[tuple[str, bool]] = None,
+        limit: Optional[int] = None,
+        offset: Optional[int] = None,
+    ) -> List[Dict[str, Any]]:
+        query = self.client.table(table).select(columns)
+        if eq:
+            query = query.eq(eq[0], eq[1])
+        if order:
+            query = query.order(order[0], desc=order[1])
+        if limit:
+            query = query.limit(limit)
+        if offset:
+            query = query.offset(offset)
+        result = await asyncio.to_thread(query.execute)
+        return result.data if result else []
+
+    async def select_in(
+        self, table: str, column: str, values: List[Any],
+        columns: str = "*",
+    ) -> List[Dict[str, Any]]:
+        query = self.client.table(table).select(columns).in_(column, values)
+        result = await asyncio.to_thread(query.execute)
+        return result.data if result else []
+
+    async def insert(
+        self, table: str, data: Dict[str, Any],
+        returning: str = "representation",
+    ) -> Optional[Dict[str, Any]]:
+        query = self.client.table(table).insert(data, returning=returning)
+        result = await asyncio.to_thread(query.execute)
+        if result and result.data:
+            return result.data[0]
+        return None
+
+    async def update(
+        self, table: str, column: str, value: Any,
+        data: Dict[str, Any],
+    ) -> List[Dict[str, Any]]:
+        query = self.client.table(table).update(data).eq(column, value)
+        result = await asyncio.to_thread(query.execute)
+        return result.data if result else []
+
+    async def delete(
+        self, table: str, column: str, value: Any,
+    ) -> List[Dict[str, Any]]:
+        query = self.client.table(table).delete().eq(column, value)
+        result = await asyncio.to_thread(query.execute)
+        return result.data if result else []
+
+    async def upsert(
+        self, table: str, data: Dict[str, Any],
+        on_conflict: str = "id",
+    ) -> Optional[Dict[str, Any]]:
+        query = self.client.table(table).upsert(data, on_conflict=on_conflict)
+        result = await asyncio.to_thread(query.execute)
+        if result and result.data:
+            return result.data[0]
+        return None
+
+    async def execute_sql(self, sql: str) -> Any:
+        result = await asyncio.to_thread(
+            self.client.rpc,
+            "exec_sql",
+            {"sql": sql},
+        )
+        return result
+
+    async def find_one(
+        self, table: str, column: str, value: Any,
+        columns: str = "*",
+    ) -> Optional[Dict[str, Any]]:
+        rows = await self.select(table, columns=columns, eq=(column, value), limit=1)
+        return rows[0] if rows else None
+
+    async def find_all(
+        self, table: str, column: str, value: Any,
+        columns: str = "*",
+    ) -> List[Dict[str, Any]]:
+        return await self.select(table, columns=columns, eq=(column, value))
+
+    async def count(
+        self, table: str, column: str, value: Any,
+    ) -> int:
+        query = self.client.table(table).select("*", count="exact").eq(column, value)
+        result = await asyncio.to_thread(query.execute)
+        return result.count if hasattr(result, "count") and result.count else 0
+
+
+_client_instance: Optional[SupabaseClient] = None
+
+
+def get_supabase_client() -> Optional[SupabaseClient]:
+    global _client_instance
+    return _client_instance
+
+
+def set_supabase_client(client: SupabaseClient):
+    global _client_instance
+    _client_instance = client
+
+
+async def create_supabase_client() -> SupabaseClient:
+    settings = get_settings()
+    url = settings.supabase_url
+    key = settings.supabase_service_role_key
+    if not url or not key:
+        logger.warning("Supabase not configured (SUPABASE_URL or SUPABASE_SERVICE_ROLE_KEY missing)")
+        return None
+    client = SupabaseClient(url, key)
+    await client.initialize()
+    set_supabase_client(client)
+    return client

app/services/supabase/repositories.py

@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from app.core.auth.models import RefreshSession, User, _uuid
+from app.core.auth.models import _utcnow as _now
+from app.core.vector_store.models import VectorStoreIndex
+from app.services.supabase.client import SupabaseClient
+
+logger = logging.getLogger(__name__)
+
+
+def _serialize_dt(dt: Optional[datetime]) -> Optional[str]:
+    if dt is None:
+        return None
+    return dt.isoformat()
+
+
+def _deserialize_dt(val) -> Optional[datetime]:
+    if val is None:
+        return None
+    if isinstance(val, datetime):
+        return val
+    if isinstance(val, str):
+        return datetime.fromisoformat(val)
+    return val
+
+
+def _row_to_user(row: dict) -> User:
+    return User(
+        id=row["id"],
+        email=row["email"],
+        username=row.get("username"),
+        full_name=row.get("full_name"),
+        password_hash=row["password_hash"],
+        is_active=bool(row["is_active"]),
+        is_verified=bool(row["is_verified"]),
+        failed_login_attempts=row["failed_login_attempts"],
+        locked_until=_deserialize_dt(row.get("locked_until")),
+        last_login=_deserialize_dt(row.get("last_login")),
+        password_changed_at=_deserialize_dt(row.get("password_changed_at")),
+        created_at=_deserialize_dt(row["created_at"]),
+        updated_at=_deserialize_dt(row["updated_at"]),
+        deleted_at=_deserialize_dt(row.get("deleted_at")),
+    )
+
+
+def _row_to_refresh_session(row: dict) -> RefreshSession:
+    return RefreshSession(
+        id=row["id"],
+        user_id=row["user_id"],
+        token_key=row["token_key"],
+        token_hash=row["token_hash"],
+        device_info=row.get("device_info"),
+        ip_address=row.get("ip_address"),
+        expires_at=_deserialize_dt(row["expires_at"]),
+        revoked_at=_deserialize_dt(row.get("revoked_at")),
+        created_at=_deserialize_dt(row["created_at"]),
+        updated_at=_deserialize_dt(row["updated_at"]),
+    )
+
+
+class AuthRepository:
+    def __init__(self, client: SupabaseClient):
+        self._client = client
+
+    async def find_user_by_email(self, email: str) -> Optional[User]:
+        row = await self._client.find_one("users", "email", email)
+        if row is None or row.get("deleted_at") is not None:
+            return None
+        return _row_to_user(row)
+
+    async def find_user_by_id(self, user_id: str) -> Optional[User]:
+        row = await self._client.find_one("users", "id", user_id)
+        if row is None:
+            return None
+        return _row_to_user(row)
+
+    async def find_user_by_username(self, username: str) -> Optional[User]:
+        row = await self._client.find_one("users", "username", username)
+        if row is None or row.get("deleted_at") is not None:
+            return None
+        return _row_to_user(row)
+
+    async def create_user(self, user: User) -> User:
+        data = {
+            "id": user.id,
+            "email": user.email,
+            "username": user.username,
+            "full_name": user.full_name,
+            "password_hash": user.password_hash,
+            "is_active": user.is_active,
+            "is_verified": user.is_verified,
+            "failed_login_attempts": user.failed_login_attempts,
+            "locked_until": _serialize_dt(user.locked_until),
+            "last_login": _serialize_dt(user.last_login),
+            "password_changed_at": _serialize_dt(user.password_changed_at),
+            "created_at": _serialize_dt(user.created_at),
+            "updated_at": _serialize_dt(user.updated_at),
+            "deleted_at": _serialize_dt(user.deleted_at),
+        }
+        await self._client.insert("users", data)
+        return user
+
+    async def update_user(self, user_id: str, updates: Dict[str, Any]) -> None:
+        data = {}
+        for k, v in updates.items():
+            if isinstance(v, datetime):
+                data[k] = _serialize_dt(v)
+            else:
+                data[k] = v
+        data["updated_at"] = _serialize_dt(_now())
+        await self._client.update("users", "id", user_id, data)
+
+    async def get_role_by_name(self, name: str) -> Optional[Dict[str, Any]]:
+        return await self._client.find_one("roles", "name", name)
+
+    async def assign_role(self, user_id: str, role_id: str) -> None:
+        await self._client.insert("user_roles", {
+            "user_id": user_id,
+            "role_id": role_id,
+        })
+
+    async def get_user_roles(self, user_id: str) -> List[str]:
+        rows = await self._client.select(
+            "roles",
+            columns="name",
+            eq=("user_id", user_id),
+        )
+        return [r["name"] for r in rows if "name" in r]
+
+    async def get_user_permissions(self, user_id: str) -> List[str]:
+        rows = await self._client.select_in(
+            "permissions",
+            "id",
+            [],
+            columns="code",
+        )
+        return [r["code"] for r in rows if "code" in r]
+
+    async def create_refresh_session(self, session: RefreshSession) -> None:
+        data = {
+            "id": session.id,
+            "user_id": session.user_id,
+            "token_key": session.token_key,
+            "token_hash": session.token_hash,
+            "device_info": session.device_info,
+            "ip_address": session.ip_address,
+            "expires_at": _serialize_dt(session.expires_at),
+            "revoked_at": _serialize_dt(session.revoked_at),
+            "created_at": _serialize_dt(session.created_at),
+            "updated_at": _serialize_dt(session.updated_at),
+        }
+        await self._client.insert("refresh_sessions", data)
+
+    async def find_session_by_token_key(self, token_key: str) -> Optional[RefreshSession]:
+        row = await self._client.find_one("refresh_sessions", "token_key", token_key)
+        if row is None:
+            return None
+        return _row_to_refresh_session(row)
+
+    async def find_valid_session(self, token_key: str) -> Optional[RefreshSession]:
+        rows = await self._client.select(
+            "refresh_sessions",
+            eq=("token_key", token_key),
+        )
+        now = _serialize_dt(_now())
+        for row in rows:
+            if row.get("revoked_at") is None and row.get("expires_at", "") > now:
+                return _row_to_refresh_session(row)
+        return None
+
+    async def revoke_session(self, session_id: str) -> None:
+        now = _serialize_dt(_now())
+        await self._client.update("refresh_sessions", "id", session_id, {"revoked_at": now})
+
+    async def revoke_all_user_sessions(self, user_id: str) -> None:
+        now = _serialize_dt(_now())
+        rows = await self._client.select(
+            "refresh_sessions",
+            eq=("user_id", user_id),
+        )
+        for row in rows:
+            if row.get("revoked_at") is None:
+                await self._client.update(
+                    "refresh_sessions", "id", row["id"], {"revoked_at": now},
+                )
+
+    async def find_user_sessions(self, user_id: str) -> List[RefreshSession]:
+        rows = await self._client.select(
+            "refresh_sessions",
+            eq=("user_id", user_id),
+            order=("created_at", True),
+        )
+        return [_row_to_refresh_session(r) for r in rows if r.get("revoked_at") is None]
+
+    async def cleanup_expired_sessions(self) -> int:
+        rows = await self._client.select("refresh_sessions")
+        count = 0
+        now = _serialize_dt(_now())
+        for row in rows:
+            if row.get("expires_at", "") < now and row.get("revoked_at") is None:
+                await self._client.update(
+                    "refresh_sessions", "id", row["id"], {"revoked_at": now},
+                )
+                count += 1
+        return count
+
+    async def get_user_permission_codes(self, user_id: str) -> List[str]:
+        return await self._client.select_in(
+            "permissions",
+            "id",
+            [],
+            columns="code",
+        )
+
+
+class VectorStoreRepository:
+    def __init__(self, client: SupabaseClient):
+        self._client = client
+
+    async def find_all(self) -> List[VectorStoreIndex]:
+        rows = await self._client.select("vector_store_index", order=("created_at", False))
+        return [VectorStoreIndex.from_dict(r) for r in rows]
+
+    async def find_by_id(self, store_id: str) -> Optional[VectorStoreIndex]:
+        row = await self._client.find_one("vector_store_index", "store_id", store_id)
+        if row is None:
+            return None
+        return VectorStoreIndex.from_dict(row)
+
+    async def upsert(self, index: VectorStoreIndex) -> None:
+        data = {
+            "store_id": index.store_id,
+            "name": index.name,
+            "path": index.path,
+            "description": index.description,
+            "metadata_json": index.metadata_json,
+            "created_at": index.created_at,
+        }
+        existing = await self._client.find_one("vector_store_index", "store_id", index.store_id)
+        if existing:
+            await self._client.update("vector_store_index", "store_id", index.store_id, {
+                "name": index.name,
+                "description": index.description,
+                "metadata_json": index.metadata_json,
+            })
+        else:
+            await self._client.insert("vector_store_index", data)
+
+    async def delete(self, store_id: str) -> None:
+        await self._client.delete("vector_store_index", "store_id", store_id)

app/services/url_shortener_service.py

@@ -20,7 +20,7 @@ from urllib.parse import urlparse
 
 from app.config import get_settings
 from app.core.logger import get_logger
-from app.core.tidb_manager import TiDBManager, get_tidb_manager
+from app.services.supabase import SupabaseClient, get_supabase_client
 
 logger = get_logger(__name__)
 
@@ -313,115 +313,22 @@ def _fire_click_webhook(webhook_url: str, short_code: str, long_url: str,
 
 
 # ---------------------------------------------------------------------------
-# TiDB-backed Storage
+# Supabase-backed Storage
 # ---------------------------------------------------------------------------
 
-US_TABLES: dict[str, str] = {
-    "url_shortener_owners": """
-        CREATE TABLE IF NOT EXISTS url_shortener_owners (
-            owner_id      VARCHAR(64) PRIMARY KEY,
-            name          VARCHAR(255) NOT NULL,
-            api_key_hash  VARCHAR(128) NOT NULL,
-            plan          VARCHAR(32) NOT NULL DEFAULT 'free',
-            created_at    VARCHAR(64) NOT NULL,
-            UNIQUE KEY uk_uso_api_key_hash (api_key_hash)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "url_shortener_links": """
-        CREATE TABLE IF NOT EXISTS url_shortener_links (
-            short_code       VARCHAR(32) PRIMARY KEY,
-            long_url         TEXT NOT NULL,
-            owner_id         VARCHAR(64) NOT NULL,
-            created_at       VARCHAR(64) NOT NULL,
-            expires_at       VARCHAR(64),
-            max_clicks       INT,
-            click_count      INT NOT NULL DEFAULT 0,
-            last_accessed_at VARCHAR(64),
-            is_active        TINYINT(1) NOT NULL DEFAULT 1,
-            tags             VARCHAR(512),
-            note             TEXT,
-            password_hash    VARCHAR(128),
-            utm_source       VARCHAR(512),
-            utm_medium       VARCHAR(512),
-            utm_campaign     VARCHAR(512),
-            campaign_id      VARCHAR(64),
-            custom_domain    VARCHAR(255),
-            fallback_url     TEXT,
-            webhook_url      TEXT,
-            geo_targeting    TEXT,
-            INDEX idx_usl_owner (owner_id),
-            INDEX idx_usl_campaign (campaign_id),
-            CONSTRAINT fk_usl_owner FOREIGN KEY (owner_id) REFERENCES url_shortener_owners(owner_id)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "url_shortener_clicks": """
-        CREATE TABLE IF NOT EXISTS url_shortener_clicks (
-            id          BIGINT AUTO_INCREMENT PRIMARY KEY,
-            short_code  VARCHAR(32) NOT NULL,
-            referrer    TEXT,
-            user_agent  TEXT,
-            ip_address  VARCHAR(45),
-            country     VARCHAR(64),
-            browser     VARCHAR(64),
-            device      VARCHAR(64),
-            os          VARCHAR(64),
-            clicked_at  VARCHAR(64) NOT NULL,
-            INDEX idx_usc_code (short_code),
-            INDEX idx_usc_clicked_at (clicked_at),
-            CONSTRAINT fk_usc_link FOREIGN KEY (short_code) REFERENCES url_shortener_links(short_code)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "url_shortener_audit_log": """
-        CREATE TABLE IF NOT EXISTS url_shortener_audit_log (
-            id         BIGINT AUTO_INCREMENT PRIMARY KEY,
-            timestamp  VARCHAR(64) NOT NULL,
-            owner_id   VARCHAR(64),
-            action     VARCHAR(64) NOT NULL,
-            short_code VARCHAR(32),
-            detail     TEXT
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-    "url_shortener_campaigns": """
-        CREATE TABLE IF NOT EXISTS url_shortener_campaigns (
-            campaign_id  VARCHAR(64) PRIMARY KEY,
-            owner_id     VARCHAR(64) NOT NULL,
-            name         VARCHAR(255) NOT NULL,
-            description  TEXT,
-            created_at   VARCHAR(64) NOT NULL,
-            is_active    TINYINT(1) NOT NULL DEFAULT 1,
-            INDEX idx_uscamp_owner (owner_id),
-            CONSTRAINT fk_uscamp_owner FOREIGN KEY (owner_id) REFERENCES url_shortener_owners(owner_id)
-        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-    """,
-}
-
-
-def _run_async(coro):
-    try:
-        loop = asyncio.get_running_loop()
-        if loop.is_running():
-            import concurrent.futures
-            with concurrent.futures.ThreadPoolExecutor(max_workers=1):
-                future = asyncio.run_coroutine_threadsafe(coro, loop)
-                return future.result()
-    except RuntimeError:
-        pass
-    return asyncio.run(coro)
-
 
 class Storage:
     def __init__(self):
         self._lock = threading.RLock()
 
-    def _get_mgr(self) -> TiDBManager:
-        mgr = get_tidb_manager()
-        if mgr is None:
-            raise RuntimeError("TiDB manager not initialized")
-        return mgr
+    def _get_client(self) -> SupabaseClient:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase client not initialized")
+        return client
 
     def _init_schema(self):
-        mgr = self._get_mgr()
-        _run_async(mgr.ensure_tables(US_TABLES))
+        logger.info("URL shortener schema managed via Supabase (DDL in /ddl directory)")
 
     def close(self):
         pass
@@ -430,288 +337,263 @@ class Storage:
 
     def create_owner(self, owner_id: str, name: str, api_key_hash: str, plan: str = "free"):
         with self._lock:
-            _run_async(self._get_mgr().execute(
-                "INSERT INTO url_shortener_owners (owner_id, name, api_key_hash, plan, created_at) "
-                "VALUES (%s, %s, %s, %s, %s)",
-                (owner_id, name, api_key_hash, plan, _now_iso()),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().insert("url_shortener_owners", {
+                "owner_id": owner_id,
+                "name": name,
+                "api_key_hash": api_key_hash,
+                "plan": plan,
+                "created_at": _now_iso(),
+            }))
 
     def get_owner_by_key_hash(self, api_key_hash: str) -> Optional[dict]:
-        return _run_async(self._get_mgr().fetchone_all(
-            "SELECT * FROM url_shortener_owners WHERE api_key_hash = %s",
-            (api_key_hash,),
-        ))
+        import asyncio
+        return asyncio.run(self._get_client().find_one("url_shortener_owners", "api_key_hash", api_key_hash))
 
     def get_owner(self, owner_id: str) -> Optional[dict]:
-        return _run_async(self._get_mgr().fetchone_all(
-            "SELECT * FROM url_shortener_owners WHERE owner_id = %s",
-            (owner_id,),
-        ))
+        import asyncio
+        return asyncio.run(self._get_client().find_one("url_shortener_owners", "owner_id", owner_id))
 
     def update_owner_plan(self, owner_id: str, plan: str):
         with self._lock:
-            _run_async(self._get_mgr().execute(
-                "UPDATE url_shortener_owners SET plan = %s WHERE owner_id = %s",
-                (plan, owner_id),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().update("url_shortener_owners", "owner_id", owner_id, {"plan": plan}))
 
     # ---------- Links ----------
 
     def code_exists(self, short_code: str) -> bool:
-        row = _run_async(self._get_mgr().fetchone_all(
-            "SELECT 1 AS chk FROM url_shortener_links WHERE short_code = %s",
-            (short_code,),
-        ))
+        import asyncio
+        row = asyncio.run(self._get_client().find_one("url_shortener_links", "short_code", short_code, columns="short_code"))
         return row is not None
 
     def insert_link(self, **kwargs):
         with self._lock:
-            _run_async(self._get_mgr().insert_and_get_id("""
-                INSERT INTO url_shortener_links
-                    (short_code, long_url, owner_id, created_at, expires_at,
-                     max_clicks, click_count, last_accessed_at, is_active, tags, note,
-                     password_hash, utm_source, utm_medium, utm_campaign,
-                     campaign_id, custom_domain, fallback_url, webhook_url, geo_targeting)
-                VALUES
-                    (%(short_code)s, %(long_url)s, %(owner_id)s, %(created_at)s, %(expires_at)s,
-                     %(max_clicks)s, 0, NULL, 1, %(tags)s, %(note)s,
-                     %(password_hash)s, %(utm_source)s, %(utm_medium)s, %(utm_campaign)s,
-                     %(campaign_id)s, %(custom_domain)s, %(fallback_url)s, %(webhook_url)s,
-                     %(geo_targeting)s)
-            """, kwargs))
+            data = {
+                "short_code": kwargs["short_code"],
+                "long_url": kwargs["long_url"],
+                "owner_id": kwargs["owner_id"],
+                "created_at": kwargs.get("created_at", _now_iso()),
+                "expires_at": kwargs.get("expires_at"),
+                "max_clicks": kwargs.get("max_clicks"),
+                "click_count": 0,
+                "is_active": 1,
+                "tags": kwargs.get("tags"),
+                "note": kwargs.get("note"),
+                "password_hash": kwargs.get("password_hash"),
+                "utm_source": kwargs.get("utm_source"),
+                "utm_medium": kwargs.get("utm_medium"),
+                "utm_campaign": kwargs.get("utm_campaign"),
+                "campaign_id": kwargs.get("campaign_id"),
+                "custom_domain": kwargs.get("custom_domain"),
+                "fallback_url": kwargs.get("fallback_url"),
+                "webhook_url": kwargs.get("webhook_url"),
+                "geo_targeting": kwargs.get("geo_targeting"),
+            }
+            import asyncio
+            asyncio.run(self._get_client().insert("url_shortener_links", data))
 
     def update_link(self, short_code: str, **updates):
         with self._lock:
-            sets = ", ".join(f"{k} = %s" for k in updates)
-            vals = list(updates.values()) + [short_code]
-            _run_async(self._get_mgr().execute(
-                f"UPDATE url_shortener_links SET {sets} WHERE short_code = %s",
-                tuple(vals),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().update("url_shortener_links", "short_code", short_code, updates))
 
     def get_link(self, short_code: str) -> Optional[dict]:
-        return _run_async(self._get_mgr().fetchone_all(
-            "SELECT * FROM url_shortener_links WHERE short_code = %s",
-            (short_code,),
-        ))
+        import asyncio
+        return asyncio.run(self._get_client().find_one("url_shortener_links", "short_code", short_code))
 
     def list_links_for_owner(self, owner_id: str) -> List[dict]:
-        return _run_async(self._get_mgr().fetchall_all(
-            "SELECT * FROM url_shortener_links WHERE owner_id = %s ORDER BY created_at DESC",
-            (owner_id,),
+        import asyncio
+        return asyncio.run(self._get_client().select(
+            "url_shortener_links", eq=("owner_id", owner_id), order=("created_at", True),
         ))
 
     def deactivate_link(self, short_code: str):
         with self._lock:
-            _run_async(self._get_mgr().execute(
-                "UPDATE url_shortener_links SET is_active = 0 WHERE short_code = %s",
-                (short_code,),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().update("url_shortener_links", "short_code", short_code, {"is_active": 0}))
 
     def record_click(self, short_code: str, referrer: Optional[str] = None,
                      user_agent: Optional[str] = None, ip_address: Optional[str] = None):
         now = _now_iso()
         parsed = _parse_user_agent(user_agent or "")
         with self._lock:
-            _run_async(self._get_mgr().execute(
-                "INSERT INTO url_shortener_clicks "
-                "(short_code, referrer, user_agent, ip_address, country, browser, device, os, clicked_at) "
-                "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
-                (short_code, referrer, user_agent, ip_address, None,
-                 parsed["browser"], parsed["device"], parsed["os"], now),
-            ))
-            _run_async(self._get_mgr().execute(
-                "UPDATE url_shortener_links SET click_count = click_count + 1, last_accessed_at = %s "
-                "WHERE short_code = %s",
-                (now, short_code),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().insert("url_shortener_clicks", {
+                "short_code": short_code,
+                "referrer": referrer,
+                "user_agent": user_agent,
+                "ip_address": ip_address,
+                "country": None,
+                "browser": parsed["browser"],
+                "device": parsed["device"],
+                "os": parsed["os"],
+                "clicked_at": now,
+            }))
+            link = asyncio.run(self._get_client().find_one("url_shortener_links", "short_code", short_code))
+            if link:
+                current_clicks = (link.get("click_count") or 0) + 1
+                asyncio.run(self._get_client().update(
+                    "url_shortener_links", "short_code", short_code,
+                    {"click_count": current_clicks, "last_accessed_at": now},
+                ))
 
     def get_click_analytics(self, short_code: str) -> Dict[str, Any]:
-        mgr = self._get_mgr()
-        total = _run_async(mgr.fetchone_all(
-            "SELECT COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s",
-            (short_code,),
-        ))
-        total_val = total["cnt"] if total else 0
-        browsers_rows = _run_async(mgr.fetchall_all(
-            "SELECT browser, COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s "
-            "AND browser IS NOT NULL GROUP BY browser ORDER BY cnt DESC",
-            (short_code,),
-        ))
-        devices_rows = _run_async(mgr.fetchall_all(
-            "SELECT device, COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s "
-            "AND device IS NOT NULL GROUP BY device ORDER BY cnt DESC",
-            (short_code,),
-        ))
-        referrers_rows = _run_async(mgr.fetchall_all(
-            "SELECT COALESCE(NULLIF(referrer, ''), '(direct)') AS ref, COUNT(*) AS n "
-            "FROM url_shortener_clicks WHERE short_code = %s "
-            "GROUP BY ref ORDER BY n DESC LIMIT 10",
-            (short_code,),
-        ))
-        os_rows = _run_async(mgr.fetchall_all(
-            "SELECT os, COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s "
-            "AND os IS NOT NULL GROUP BY os ORDER BY cnt DESC",
-            (short_code,),
-        ))
-        recent = _run_async(mgr.fetchall_all(
-            "SELECT clicked_at FROM url_shortener_clicks WHERE short_code = %s "
-            "ORDER BY clicked_at DESC LIMIT 50",
-            (short_code,),
-        ))
+        import asyncio
+        client = self._get_client()
+        all_clicks = asyncio.run(client.select("url_shortener_clicks", eq=("short_code", short_code)))
+        total_val = len(all_clicks)
+        browsers: Dict[str, int] = {}
+        devices: Dict[str, int] = {}
+        referrers: Dict[str, int] = {}
+        os_data: Dict[str, int] = {}
+        recent = []
+        for c in all_clicks:
+            b = c.get("browser")
+            if b:
+                browsers[b] = browsers.get(b, 0) + 1
+            d = c.get("device")
+            if d:
+                devices[d] = devices.get(d, 0) + 1
+            o = c.get("os")
+            if o:
+                os_data[o] = os_data.get(o, 0) + 1
+            ref = c.get("referrer") or "(direct)"
+            referrers[ref] = referrers.get(ref, 0) + 1
+            recent.append(c.get("clicked_at"))
+        top_referrers = dict(sorted(referrers.items(), key=lambda x: -x[1])[:10])
         return {
             "total_clicks": total_val,
-            "browsers": {r["browser"]: r["cnt"] for r in browsers_rows},
-            "devices": {r["device"]: r["cnt"] for r in devices_rows},
-            "operating_systems": {r["os"]: r["cnt"] for r in os_rows},
-            "top_referrers": {r["ref"]: r["n"] for r in referrers_rows},
-            "recent_clicks": [r["clicked_at"] for r in recent],
+            "browsers": browsers,
+            "devices": devices,
+            "operating_systems": os_data,
+            "top_referrers": top_referrers,
+            "recent_clicks": recent[:50],
         }
 
     def get_link_count_for_owner(self, owner_id: str) -> int:
-        row = _run_async(self._get_mgr().fetchone_all(
-            "SELECT COUNT(*) AS cnt FROM url_shortener_links WHERE owner_id = %s",
-            (owner_id,),
-        ))
-        return row["cnt"] if row else 0
+        import asyncio
+        links = asyncio.run(self._get_client().select("url_shortener_links", eq=("owner_id", owner_id)))
+        return len(links)
 
     def audit(self, owner_id: Optional[str], action: str, short_code: Optional[str], detail: str = ""):
         with self._lock:
-            _run_async(self._get_mgr().execute(
-                "INSERT INTO url_shortener_audit_log (timestamp, owner_id, action, short_code, detail) "
-                "VALUES (%s, %s, %s, %s, %s)",
-                (_now_iso(), owner_id, action, short_code, detail),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().insert("url_shortener_audit_log", {
+                "timestamp": _now_iso(),
+                "owner_id": owner_id,
+                "action": action,
+                "short_code": short_code,
+                "detail": detail,
+            }))
 
     def export_links_csv(self, owner_id: str) -> str:
-        rows = _run_async(self._get_mgr().fetchall_all(
-            "SELECT short_code, long_url, created_at, click_count, is_active, tags, note "
-            "FROM url_shortener_links WHERE owner_id = %s ORDER BY created_at DESC",
-            (owner_id,),
+        import asyncio
+        rows = asyncio.run(self._get_client().select(
+            "url_shortener_links", eq=("owner_id", owner_id), order=("created_at", True),
         ))
         buf = io.StringIO()
         w = csv.writer(buf)
         w.writerow(["short_code", "long_url", "created_at", "click_count", "is_active", "tags", "note"])
         for r in rows:
             w.writerow([r["short_code"], r["long_url"], r["created_at"],
-                       r["click_count"], r["is_active"], r["tags"] or "", r["note"] or ""])
+                       r["click_count"], r["is_active"], r.get("tags") or "", r.get("note") or ""])
         return buf.getvalue()
 
     def export_clicks_csv(self, short_code: str) -> str:
-        rows = _run_async(self._get_mgr().fetchall_all(
-            "SELECT id, referrer, user_agent, ip_address, browser, device, os, clicked_at "
-            "FROM url_shortener_clicks WHERE short_code = %s ORDER BY clicked_at DESC",
-            (short_code,),
+        import asyncio
+        rows = asyncio.run(self._get_client().select(
+            "url_shortener_clicks", eq=("short_code", short_code), order=("clicked_at", True),
         ))
         buf = io.StringIO()
         w = csv.writer(buf)
         w.writerow(["id", "referrer", "user_agent", "ip_address", "browser", "device", "os", "clicked_at"])
         for r in rows:
-            w.writerow([r["id"], r["referrer"], r["user_agent"], r["ip_address"],
-                       r["browser"], r["device"], r["os"], r["clicked_at"]])
+            w.writerow([r["id"], r.get("referrer"), r.get("user_agent"), r.get("ip_address"),
+                       r.get("browser"), r.get("device"), r.get("os"), r.get("clicked_at")])
         return buf.getvalue()
 
     def list_links_by_campaign(self, campaign_id: str) -> List[dict]:
-        return _run_async(self._get_mgr().fetchall_all(
-            "SELECT * FROM url_shortener_links WHERE campaign_id = %s ORDER BY created_at DESC",
-            (campaign_id,),
+        import asyncio
+        return asyncio.run(self._get_client().select(
+            "url_shortener_links", eq=("campaign_id", campaign_id), order=("created_at", True),
         ))
 
     def create_campaign(self, campaign_id: str, owner_id: str, name: str, description: Optional[str] = None):
         with self._lock:
-            _run_async(self._get_mgr().execute(
-                "INSERT INTO url_shortener_campaigns (campaign_id, owner_id, name, description, created_at) "
-                "VALUES (%s, %s, %s, %s, %s)",
-                (campaign_id, owner_id, name, description, _now_iso()),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().insert("url_shortener_campaigns", {
+                "campaign_id": campaign_id,
+                "owner_id": owner_id,
+                "name": name,
+                "description": description,
+                "created_at": _now_iso(),
+            }))
 
     def get_campaign(self, campaign_id: str) -> Optional[dict]:
-        return _run_async(self._get_mgr().fetchone_all(
-            "SELECT * FROM url_shortener_campaigns WHERE campaign_id = %s",
-            (campaign_id,),
-        ))
+        import asyncio
+        return asyncio.run(self._get_client().find_one("url_shortener_campaigns", "campaign_id", campaign_id))
 
     def list_campaigns_for_owner(self, owner_id: str) -> List[dict]:
-        return _run_async(self._get_mgr().fetchall_all(
-            "SELECT * FROM url_shortener_campaigns WHERE owner_id = %s ORDER BY created_at DESC",
-            (owner_id,),
+        import asyncio
+        return asyncio.run(self._get_client().select(
+            "url_shortener_campaigns", eq=("owner_id", owner_id), order=("created_at", True),
         ))
 
     def deactivate_campaign(self, campaign_id: str):
         with self._lock:
-            _run_async(self._get_mgr().execute(
-                "UPDATE url_shortener_campaigns SET is_active = 0 WHERE campaign_id = %s",
-                (campaign_id,),
-            ))
+            import asyncio
+            asyncio.run(self._get_client().update("url_shortener_campaigns", "campaign_id", campaign_id, {"is_active": 0}))
 
     def get_campaign_link_count(self, campaign_id: str) -> int:
-        row = _run_async(self._get_mgr().fetchone_all(
-            "SELECT COUNT(*) AS cnt FROM url_shortener_links WHERE campaign_id = %s",
-            (campaign_id,),
-        ))
-        return row["cnt"] if row else 0
+        import asyncio
+        links = asyncio.run(self._get_client().select("url_shortener_links", eq=("campaign_id", campaign_id)))
+        return len(links)
 
     def get_campaign_total_clicks(self, campaign_id: str) -> int:
-        row = _run_async(self._get_mgr().fetchone_all(
-            "SELECT COALESCE(SUM(click_count), 0) AS cnt FROM url_shortener_links WHERE campaign_id = %s",
-            (campaign_id,),
-        ))
-        return row["cnt"] if row else 0
+        import asyncio
+        links = asyncio.run(self._get_client().select("url_shortener_links", eq=("campaign_id", campaign_id)))
+        return sum(l.get("click_count") or 0 for l in links)
 
     def get_campaign_analytics(self, campaign_id: str) -> Dict[str, Any]:
-        codes = _run_async(self._get_mgr().fetchall_all(
-            "SELECT short_code FROM url_shortener_links WHERE campaign_id = %s",
-            (campaign_id,),
-        ))
+        import asyncio
+        client = self._get_client()
+        codes = asyncio.run(client.select("url_shortener_links", eq=("campaign_id", campaign_id)))
         total = 0
         browsers: Dict[str, int] = {}
         devices: Dict[str, int] = {}
         os_data: Dict[str, int] = {}
-        mgr = self._get_mgr()
         for row in codes:
             code = row["short_code"]
-            cnt_row = _run_async(mgr.fetchone_all(
-                "SELECT COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s",
-                (code,),
-            ))
-            total += cnt_row["cnt"] if cnt_row else 0
-            for b in _run_async(mgr.fetchall_all(
-                "SELECT browser, COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s "
-                "AND browser IS NOT NULL GROUP BY browser", (code,),
-            )):
-                browsers[b["browser"]] = browsers.get(b["browser"], 0) + b["cnt"]
-            for d in _run_async(mgr.fetchall_all(
-                "SELECT device, COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s "
-                "AND device IS NOT NULL GROUP BY device", (code,),
-            )):
-                devices[d["device"]] = devices.get(d["device"], 0) + d["cnt"]
-            for o in _run_async(mgr.fetchall_all(
-                "SELECT os, COUNT(*) AS cnt FROM url_shortener_clicks WHERE short_code = %s "
-                "AND os IS NOT NULL GROUP BY os", (code,),
-            )):
-                os_data[o["os"]] = os_data.get(o["os"], 0) + o["cnt"]
+            clicks = asyncio.run(client.select("url_shortener_clicks", eq=("short_code", code)))
+            total += len(clicks)
+            for c in clicks:
+                b = c.get("browser")
+                if b:
+                    browsers[b] = browsers.get(b, 0) + 1
+                d = c.get("device")
+                if d:
+                    devices[d] = devices.get(d, 0) + 1
+                o = c.get("os")
+                if o:
+                    os_data[o] = os_data.get(o, 0) + 1
         return {"total_clicks": total, "browsers": browsers, "devices": devices, "operating_systems": os_data}
 
     def owner_summary(self, owner_id: str) -> Dict[str, Any]:
         row = self.get_owner(owner_id)
         if row is None:
             return {}
-        link_count = self.get_link_count_for_owner(owner_id)
-        total_clicks_row = _run_async(self._get_mgr().fetchone_all(
-            "SELECT COALESCE(SUM(click_count), 0) AS cnt FROM url_shortener_links WHERE owner_id = %s",
-            (owner_id,),
-        ))
-        total_clicks = total_clicks_row["cnt"] if total_clicks_row else 0
-        active = _run_async(self._get_mgr().fetchone_all(
-            "SELECT COUNT(*) AS cnt FROM url_shortener_links WHERE owner_id = %s AND is_active = 1",
-            (owner_id,),
-        ))
+        import asyncio
+        client = self._get_client()
+        links = asyncio.run(client.select("url_shortener_links", eq=("owner_id", owner_id)))
+        link_count = len(links)
+        total_clicks = sum(l.get("click_count") or 0 for l in links)
+        active = sum(1 for l in links if l.get("is_active"))
         return {
             "owner_id": owner_id,
             "name": row["name"],
             "plan": row["plan"],
             "total_links": link_count,
-            "active_links": active["cnt"] if active else 0,
+            "active_links": active,
             "total_clicks": total_clicks,
             "created_at": row["created_at"],
         }
@@ -721,7 +603,7 @@ class URLShortenerService:
     def __init__(self):
         self.storage = Storage()
         self.storage._init_schema()
-        logger.info("URLShortenerService initialized (TiDB)")
+        logger.info("URLShortenerService initialized (Supabase)")
 
     def close(self) -> None:
         self.storage.close()
@@ -1067,15 +949,14 @@ class URLShortenerService:
             raise LinkNotFoundError(f"No link found for code '{short_code}'")
         if row["owner_id"] != owner_id:
             raise AuthorizationError("You do not own this link")
-        mgr = get_tidb_manager()
-        if mgr is None:
+        import asyncio
+        client = get_supabase_client()
+        if client is None:
             return []
-        rows = _run_async(mgr.fetchall_all(
-            "SELECT id, referrer, user_agent, ip_address, browser, device, os, clicked_at "
-            "FROM url_shortener_clicks WHERE short_code = %s ORDER BY clicked_at DESC LIMIT %s",
-            (short_code, limit),
+        rows = asyncio.run(client.select(
+            "url_shortener_clicks", eq=("short_code", short_code),
         ))
-        return list(rows)
+        return list(rows)[:limit]
 
     def get_qr_code(self, short_code: str) -> Dict[str, str]:
         row = self.storage.get_link(short_code)

app/services/vector_store_service.py

@@ -14,7 +14,7 @@ import zvec
 
 from app.config import get_settings
 from app.core.logger import get_logger
-from app.core.tidb_manager import get_tidb_manager
+from app.services.supabase import get_supabase_client, VectorStoreRepository
 from app.services.chunking_service import chunk_text_async
 from app.services.embeddings_service import EmbeddingService
 
@@ -62,27 +62,26 @@ class VectorStoreService:
     def _get_collection(self, store_id: str) -> Optional["zvec.Collection"]:
         return self._collections.get(store_id)
 
-    # --- TiDB persistence ---
+    # --- Supabase persistence ---
 
     async def init_db(self) -> None:
-        mgr = get_tidb_manager()
-        if mgr is None:
-            logger.warning("TiDB manager not available, vector store metadata not loaded")
+        client = get_supabase_client()
+        if client is None:
+            logger.warning("Supabase not available, vector store metadata not loaded")
             return
-        rows = await mgr.fetchall_all(
-            "SELECT * FROM vector_store_index ORDER BY created_at ASC"
-        )
-        for row in rows:
+        repo = VectorStoreRepository(client)
+        indexes = await repo.find_all()
+        for index in indexes:
             record = VectorStoreRecord(
-                store_id=row["store_id"],
-                name=row["name"],
-                path=row["path"],
-                description=row.get("description", ""),
-                metadata=json.loads(row.get("metadata_json", "{}") or "{}"),
-                created_at=row["created_at"],
+                store_id=index.store_id,
+                name=index.name,
+                path=index.path,
+                description=index.description,
+                metadata=json.loads(index.metadata_json or "{}"),
+                created_at=index.created_at,
             )
             self._stores[record.store_id] = record
-            store_path = row["path"]
+            store_path = index.path
             if os.path.exists(os.path.join(store_path, "__zvec_meta")):
                 try:
                     col = zvec.open(store_path)
@@ -92,39 +91,28 @@ class VectorStoreService:
                     logger.warning("Could not open collection %s: %s", record.store_id, exc)
 
     async def _persist_store(self, record: VectorStoreRecord) -> None:
-        mgr = get_tidb_manager()
-        if mgr is None:
-            logger.error("TiDB manager not available, cannot persist store")
+        client = get_supabase_client()
+        if client is None:
+            logger.error("Supabase not available, cannot persist store")
             return
-        existing = await mgr.fetchone(
-            "SELECT store_id FROM vector_store_index WHERE store_id = %s",
-            (record.store_id,),
+        repo = VectorStoreRepository(client)
+        from app.core.vector_store.models import VectorStoreIndex
+        index = VectorStoreIndex(
+            store_id=record.store_id,
+            name=record.name,
+            path=record.path,
+            description=record.description,
+            metadata_json=json.dumps(record.metadata),
+            created_at=record.created_at,
         )
-        if existing:
-            await mgr.execute(
-                "UPDATE vector_store_index SET name = %s, description = %s, "
-                "metadata_json = %s WHERE store_id = %s",
-                (record.name, record.description, json.dumps(record.metadata), record.store_id),
-            )
-        else:
-            await mgr.execute(
-                "INSERT INTO vector_store_index (store_id, name, path, description, metadata_json, created_at) "
-                "VALUES (%s, %s, %s, %s, %s, %s)",
-                (
-                    record.store_id, record.name, record.path,
-                    record.description, json.dumps(record.metadata),
-                    record.created_at,
-                ),
-            )
+        await repo.upsert(index)
 
     async def _remove_persisted_store(self, store_id: str) -> None:
-        mgr = get_tidb_manager()
-        if mgr is None:
+        client = get_supabase_client()
+        if client is None:
             return
-        await mgr.execute(
-            "DELETE FROM vector_store_index WHERE store_id = %s",
-            (store_id,),
-        )
+        repo = VectorStoreRepository(client)
+        await repo.delete(store_id)
 
     # --- Synchronous helpers (run in thread pool) ---
 
@@ -358,7 +346,6 @@ class VectorStoreService:
             "store_id": store_id,
             "name": record.name,
             "description": record.description,
-            "app_id": settings.application_id or store_id,
             "embedding_dimension": _EMBEDDING_DIM,
             "document_count": doc_count,
             "created_at": record.created_at,
@@ -376,7 +363,7 @@ class VectorStoreService:
         name: str,
         description: str = "",
         metadata: Optional[Dict[str, Any]] = None,
-    ) -> Tuple[str, str]:
+    ) -> str:
         store_id = str(uuid.uuid4())
         store_path = self._store_path(store_id)
 
@@ -394,9 +381,8 @@ class VectorStoreService:
         self._stores[store_id] = record
         await self._persist_store(record)
 
-        app_id = settings.application_id or store_id
-        logger.info("Created vector store: %s (name=%s, app_id=%s)", store_id, name, app_id)
-        return store_id, app_id
+        logger.info("Created vector store: %s (name=%s)", store_id, name)
+        return store_id
 
     def list_stores(self) -> List[VectorStoreRecord]:
         return list(self._stores.values())

pyproject.toml

@@ -32,6 +32,7 @@ dependencies = [
     "spacy>=3.7.0",
     "phonenumbers>=8.13.0",
     "jsonschema>=4.21.0",
+    "supabase>=2.0.0",
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -39,7 +39,7 @@ email-validator>=2.1.0
 slowapi>=0.1.9
 
 # Async database drivers
-aiomysql>=0.3.2
+supabase>=2.0.0
 asyncpg>=0.31.0
 motor>=3.7.1