feat: add endpoint for retrieving user space credits and implement caching mechanism in UserService

src/langbot/pkg/api/http/controller/groups/user.py

@@ -158,6 +158,12 @@ class UserRouterGroup(group.RouterGroup):
                 }
             )
 
+        @self.route('/space-credits', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(user_email: str) -> str:
+            """Get Space credits balance for current user"""
+            credits = await self.ap.user_service.get_space_credits(user_email)
+            return self.success(data={'credits': credits})
+
         @self.route('/account-info', methods=['GET'], auth_type=group.AuthType.NONE)
         async def _() -> str:
             """Get account info for login page (account type and has_password)"""

src/langbot/pkg/api/http/service/user.py

@@ -16,10 +16,12 @@ from ....utils import constants
 class UserService:
     ap: app.Application
     _create_user_lock: asyncio.Lock
+    _space_credits_cache: typing.Dict[str, typing.Tuple[int, float]]  # {user_email: (credits, timestamp)}
 
     def __init__(self, ap: app.Application) -> None:
         self.ap = ap
         self._create_user_lock = asyncio.Lock()
+        self._space_credits_cache = {}
 
     def _get_space_config(self) -> typing.Dict[str, str]:
         """Get Space configuration from config file"""
@@ -178,6 +180,30 @@ class UserService:
                     raise ValueError(f'Failed to get user info: {data.get("msg")}')
                 return data.get('data', {})
 
+    async def get_space_credits(self, user_email: str, force_refresh: bool = False) -> int | None:
+        """Get Space credits for user with caching (60s TTL)"""
+        import time
+
+        cache_ttl = 60
+
+        if not force_refresh and user_email in self._space_credits_cache:
+            credits, ts = self._space_credits_cache[user_email]
+            if time.time() - ts < cache_ttl:
+                return credits
+
+        user_obj = await self.get_user_by_email(user_email)
+        if not user_obj or user_obj.account_type != 'space' or not user_obj.space_access_token:
+            return None
+
+        try:
+            info = await self.get_space_user_info(user_obj.space_access_token)
+            credits = info.get('credits')
+            if credits is not None:
+                self._space_credits_cache[user_email] = (credits, time.time())
+            return credits
+        except Exception:
+            return self._space_credits_cache.get(user_email, (None, 0))[0]
+
     async def refresh_space_token(self, refresh_token: str) -> typing.Dict:
         """Refresh Space access token"""
         space_config = self._get_space_config()