feat(graphql): implement secure session refresh mutation with token rotation

app/graphql/mutations.py

@@ -1,18 +1,103 @@
+import os
 import strawberry
-from auth.tokens import create_jwt_token, create_refresh_token
+from strawberry.types import Info
+from dotenv import load_dotenv
+from urllib.parse import quote
+import json
+        
+from app.graphql.context import CustomContext
+from auth.tokens import decode_token, create_jwt_token, create_refresh_token
+from app.utils.format import get_image_url
 
-@strawberry.type
+load_dotenv()
-class AuthResponse:
+
-    access_token: str
+ENV_MODE = os.getenv("ENV_MODE")
-    refresh_token: str
+IS_PROD = ENV_MODE == "prod"
-    token_type: str
+ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))
+ACCESS_TOKEN_MAX_AGE = ACCESS_TOKEN_EXPIRE_MINUTES * 60
 
 @strawberry.type
 class Mutation:
     @strawberry.mutation
-    def logout(self) -> str:
+    def logout(self, info: Info[CustomContext, None]) -> str:
+        """退出登录：清除浏览器的 HttpOnly Cookies"""
+        response = info.context.response
+        
+        # 让浏览器立刻过期这些 cookie
+        response.delete_cookie("access_token")
+        response.delete_cookie("refresh_token")
+        response.delete_cookie("user_info")
+        
         return "ok"
     
-    # 提示：像 Google OAuth 这种涉及第三方重定向的流程，
+    @strawberry.mutation
-    # 依然保留现有的 RESTful (/login/google 和 /auth/callback) 会更简单。
+    async def refresh_session(self, info: Info[CustomContext, None]) -> str:
-    # 登录成功后重定向回前端，前端拿到 token，后续所有数据交互全部走 GraphQL。
+        """
+        刷新 Session：如果 access_token 过期了，前端可以调用这个接口用 refresh_token 换新的 access_token。
+        同时如果 refresh_token 已经过半寿命了，这个接口也会顺便轮转一个新的 refresh_token，确保用户体验的连续性。
+        """
+        request = info.context.request
+        response = info.context.response
+        
+        # 从 Cookie 获取并校验 refresh_token
+        refresh_token = request.cookies.get("refresh_token")
+        if not refresh_token:
+            raise Exception("No refresh token found in cookies")
+            
+        try:
+            payload = decode_token(refresh_token)
+        except Exception:
+            raise Exception("Invalid or expired refresh token")
+            
+        if payload.get("type") != "refresh":
+            raise Exception("Token type mismatch")
+
+        user = await info.context.get_current_user()
+        
+        # 如果 access_token 已经彻底过期导致 Context 没捞到用户，
+        # 我们用 refresh_token 里的 sub 去 Context 的 db 里再查一遍用户
+        from app.models.user import User
+        if not user:
+            user_id = payload.get("sub")
+            user = info.context.db.query(User).filter(User.id == user_id).first()
+        if not user:
+            raise Exception("User no longer exists")
+            
+        # 始终生成并写入新的短效 access_token
+        token_data = {"sub": str(user.id), "email": user.email}
+        new_access = create_jwt_token(token_data)
+        
+        ACCESS_TOKEN_MAX_AGE = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60")) * 60
+        response.set_cookie(
+            key="access_token", value=new_access, httponly=True,
+            max_age=ACCESS_TOKEN_MAX_AGE, samesite="lax", secure=IS_PROD,
+        )
+        
+        # 检查 refresh_token 是否已经消耗了超过一半的生命周期
+        import time
+        exp = payload.get("exp")  # 过期绝对秒数
+        iat = payload.get("iat")  # 签发绝对秒数
+        now = int(time.time())
+        
+        # 如果当前时间已经过了总寿命的一半，就触发轮转，下发新的 refresh_token
+        if (now - iat) > (exp - iat) / 2:
+            new_refresh = create_refresh_token(token_data)
+            REFRESH_TOKEN_MAX_AGE = int(os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", "7")) * 24 * 60 * 60
+            response.set_cookie(
+                key="refresh_token", value=new_refresh, httponly=True,
+                max_age=REFRESH_TOKEN_MAX_AGE, samesite="lax", secure=IS_PROD,
+            )
+
+        # 同步写入并续期前端所需的 user_info        
+        response.set_cookie(
+            key="user_info",
+            value=quote(json.dumps({
+                "email": user.email,
+                "name": user.name,
+                "avatar": get_image_url(user.avatar) if user.avatar else None
+            })),
+            max_age=ACCESS_TOKEN_MAX_AGE,
+            httponly=False
+        )
+        
+        return "ok"