"""Password reset service for VoIdeaAI.
Generates and validates reset tokens (JWT, type=reset, exp=1h).
Sends reset link via email_service when SMTP is configured.
"""
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import get_settings
from app.core.security import create_reset_token, decode_reset_token, get_password_hash
from app.models.user import User
from app.services.email_service import send_email
RESET_TOKEN_EXPIRE_HOURS = 1
async def send_reset_email(db: AsyncSession, email: str) -> tuple[bool, str]:
"""Send password reset email.
Args:
db: Database session
email: User's email address
Returns:
Tuple of (success, message)
"""
result = await db.execute(select(User).where(User.email == email))
user = result.scalar_one_or_none()
if not user:
return True, "If the email exists, a reset link has been sent."
settings = get_settings()
if not settings.smtp_host:
return False, "SMTP not configured. Password reset is unavailable."
token = create_reset_token({"sub": str(user.id)})
reset_link = f"{settings.server_external_url}/reset-password?token={token}"
html = _render_reset_email(user.display_name, reset_link)
sent = await send_email(user.email, "Сброс пароля — VoIdeaAI", html)
return sent, "If the email exists, a reset link has been sent."
async def reset_password(db: AsyncSession, token: str, new_password: str) -> tuple[bool, str]:
"""Reset password using a valid reset token.
Args:
db: Database session
token: JWT reset token
new_password: New password
Returns:
Tuple of (success, message)
"""
payload = decode_reset_token(token)
if not payload or payload.get("type") != "reset":
return False, "Invalid or expired reset token."
user_id = payload.get("sub")
if not user_id:
return False, "Invalid token payload."
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
return False, "User not found."
if not user.password_hash:
return False, "Cannot reset password for OAuth-only account."
user.password_hash = get_password_hash(new_password)
await db.commit()
return True, "Password has been reset successfully."
def _render_reset_email(username: str, reset_link: str) -> str:
"""Render password reset email HTML."""
return f"""
Сброс пароля — VoIdeaAI
Привет, {username}!
Вы запросили сброс пароля. Нажмите на ссылку ниже:
Сбросить пароль
Ссылка действует 1 час.
Если вы не запрашивали сброс — проигнорируйте это письмо.
VoIdeaAI — идеи рождаются вслух, решения приходят мгновенно!
"""