fix: upsert refresh tokens (ON CONFLICT) + periodic cleanup of expired/revoked tokens

app/cabinet/routes/auth.py

@@ -144,22 +144,28 @@ async def _store_refresh_token(
     refresh_token: str,
     device_info: str | None = None,
 ) -> None:
-    """Store refresh token hash in database."""
+    """Store refresh token hash in database using upsert to avoid duplicate key errors."""
+    from sqlalchemy.dialects.postgresql import insert as pg_insert
+
     token_hash = hashlib.sha256(refresh_token.encode()).hexdigest()
     expires_at = get_refresh_token_expires_at()
 
-    token_record = CabinetRefreshToken(
+    stmt = pg_insert(CabinetRefreshToken).values(
         user_id=user_id,
         token_hash=token_hash,
         device_info=device_info,
         expires_at=expires_at,
     )
-    db.add(token_record)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        logger.debug('Refresh token already exists (duplicate)', user_id=user_id)
+    stmt = stmt.on_conflict_do_update(
+        index_elements=['token_hash'],
+        set_={
+            'expires_at': expires_at,
+            'device_info': device_info,
+            'revoked_at': None,
+        },
+    )
+    await db.execute(stmt)
+    await db.commit()
 
 
 async def _process_campaign_bonus(

app/services/monitoring_service.py

@@ -248,6 +248,7 @@ class MonitoringService:
                 await self._check_traffic_warnings(db)
                 await self._check_low_balance_alerts(db)
                 await self._retry_stuck_guest_purchases(db)
+                await self._cleanup_expired_refresh_tokens(db)
                 await self._cleanup_inactive_users(db)
                 await self._sync_with_remnawave(db)
 
@@ -2147,6 +2148,30 @@ class MonitoringService:
         except Exception as error:
             logger.error('Error checking low balance alerts', error=error)
 
+    async def _cleanup_expired_refresh_tokens(self, db: AsyncSession):
+        """Delete expired and revoked refresh tokens to prevent table bloat."""
+        try:
+            from sqlalchemy import delete
+
+            from app.database.models import CabinetRefreshToken
+
+            now = datetime.now(UTC)
+            # Delete tokens that are either expired or revoked more than 24h ago
+            stmt = delete(CabinetRefreshToken).where(
+                (CabinetRefreshToken.expires_at < now) | (CabinetRefreshToken.revoked_at < now - timedelta(hours=24))
+            )
+            result = await db.execute(stmt)
+            deleted = result.rowcount
+            if deleted > 0:
+                await db.commit()
+                logger.info('Cleaned up expired/revoked refresh tokens', deleted_count=deleted)
+        except Exception as error:
+            logger.error('Error cleaning up refresh tokens', error=error)
+            try:
+                await db.rollback()
+            except Exception:
+                pass
+
     async def _cleanup_inactive_users(self, db: AsyncSession):
         try:
             now = datetime.now(UTC)