feat: email login, smtp codes

db/dal/user_dal.py

@@ -24,6 +24,11 @@ from ..models import (
 REFERRAL_CODE_ALPHABET = string.ascii_uppercase + string.digits
 REFERRAL_CODE_LENGTH = 9
 MAX_REFERRAL_CODE_ATTEMPTS = 25
+MAX_EMAIL_USER_ID_ATTEMPTS = 25
+
+
+class UserMergeConflictError(ValueError):
+    pass
 
 
 def _generate_referral_code_candidate() -> str:
@@ -50,6 +55,14 @@ async def generate_unique_referral_code(session: AsyncSession) -> str:
     raise RuntimeError("Failed to generate a unique referral code after several attempts.")
 
 
+async def generate_unique_email_user_id(session: AsyncSession) -> int:
+    for _ in range(MAX_EMAIL_USER_ID_ATTEMPTS):
+        candidate = -(secrets.randbelow(9_000_000_000_000_000) + 1)
+        if not await get_user_by_id(session, candidate):
+            return candidate
+    raise RuntimeError("Failed to generate a unique email user id after several attempts.")
+
+
 async def ensure_referral_code(session: AsyncSession, user: User) -> str:
     """
     Ensure the provided user has a referral code, generating and persisting it if missing.
@@ -82,6 +95,23 @@ async def get_user_by_username(session: AsyncSession, username: str) -> Optional
     return result.scalar_one_or_none()
 
 
+async def get_user_by_email(session: AsyncSession, email: str) -> Optional[User]:
+    clean_email = (email or "").strip().lower()
+    if not clean_email:
+        return None
+    stmt = select(User).where(func.lower(User.email) == clean_email)
+    result = await session.execute(stmt)
+    return result.scalar_one_or_none()
+
+
+async def get_user_by_telegram_id(
+    session: AsyncSession, telegram_id: int
+) -> Optional[User]:
+    stmt = select(User).where(User.telegram_id == telegram_id)
+    result = await session.execute(stmt)
+    return result.scalar_one_or_none()
+
+
 async def get_user_by_panel_uuid(
     session: AsyncSession, panel_uuid: str
 ) -> Optional[User]:
@@ -135,6 +165,217 @@ async def create_user(session: AsyncSession, user_data: Dict[str, Any]) -> Tuple
     return user, created
 
 
+async def create_email_user(
+    session: AsyncSession,
+    *,
+    email: str,
+    language_code: str,
+    email_verified_at: Optional[datetime] = None,
+    referred_by_id: Optional[int] = None,
+) -> Tuple[User, bool]:
+    normalized_email = (email or "").strip().lower()
+    user_id = await generate_unique_email_user_id(session)
+    return await create_user(
+        session,
+        {
+            "user_id": user_id,
+            "email": normalized_email,
+            "email_verified_at": email_verified_at or datetime.now(timezone.utc),
+            "language_code": language_code,
+            "referred_by_id": referred_by_id,
+            "registration_date": datetime.now(timezone.utc),
+        },
+    )
+
+
+async def _has_active_panel_subscription(
+    session: AsyncSession, user_id: int, panel_user_uuid: str
+) -> bool:
+    stmt = (
+        select(Subscription.subscription_id)
+        .where(
+            Subscription.user_id == user_id,
+            Subscription.panel_user_uuid == panel_user_uuid,
+            Subscription.is_active == True,
+            Subscription.end_date > datetime.now(timezone.utc),
+        )
+        .limit(1)
+    )
+    result = await session.execute(stmt)
+    return result.scalar_one_or_none() is not None
+
+
+async def merge_users(
+    session: AsyncSession,
+    *,
+    source_user_id: int,
+    target_user_id: int,
+) -> User:
+    """Merge source user data into target user and remove the source row."""
+
+    if source_user_id == target_user_id:
+        target = await get_user_by_id(session, target_user_id)
+        if not target:
+            raise ValueError("Target user not found.")
+        return target
+
+    source = await get_user_by_id(session, source_user_id)
+    target = await get_user_by_id(session, target_user_id)
+    if not source or not target:
+        raise ValueError("Both source and target users are required for merge.")
+
+    if source.email and target.email and source.email != target.email:
+        raise UserMergeConflictError("Both accounts already have different emails.")
+    if (
+        source.telegram_id
+        and target.telegram_id
+        and int(source.telegram_id) != int(target.telegram_id)
+    ):
+        raise UserMergeConflictError("Both accounts already have different Telegram IDs.")
+
+    source_panel_uuid = source.panel_user_uuid
+    target_panel_uuid = target.panel_user_uuid
+    panel_uuid_to_move = None
+    if source_panel_uuid and not target_panel_uuid:
+        panel_uuid_to_move = source_panel_uuid
+    elif source_panel_uuid and target_panel_uuid and source_panel_uuid != target_panel_uuid:
+        source_has_active = await _has_active_panel_subscription(
+            session, source_user_id, source_panel_uuid
+        )
+        target_has_active = await _has_active_panel_subscription(
+            session, target_user_id, target_panel_uuid
+        )
+        if source_has_active and target_has_active:
+            raise UserMergeConflictError(
+                "Both accounts have active subscriptions on different panel users."
+            )
+        if source_has_active and not target_has_active:
+            panel_uuid_to_move = source_panel_uuid
+
+    email_to_move = source.email if source.email and not target.email else None
+    email_verified_at_to_move = (
+        source.email_verified_at
+        if source.email and (not target.email_verified_at or email_to_move)
+        else None
+    )
+    telegram_id_to_move = (
+        source.telegram_id if source.telegram_id and not target.telegram_id else None
+    )
+    referral_code_to_move = (
+        source.referral_code if source.referral_code and not target.referral_code else None
+    )
+
+    if email_to_move:
+        source.email = None
+    if panel_uuid_to_move:
+        source.panel_user_uuid = None
+    if telegram_id_to_move:
+        source.telegram_id = None
+    if referral_code_to_move:
+        source.referral_code = None
+    if email_to_move or panel_uuid_to_move or telegram_id_to_move or referral_code_to_move:
+        await session.flush()
+
+    if email_to_move:
+        target.email = email_to_move
+    if email_verified_at_to_move and not target.email_verified_at:
+        target.email_verified_at = email_verified_at_to_move
+    if telegram_id_to_move:
+        target.telegram_id = telegram_id_to_move
+    if panel_uuid_to_move:
+        target.panel_user_uuid = panel_uuid_to_move
+    if referral_code_to_move:
+        target.referral_code = referral_code_to_move
+
+    for attr in ("username", "first_name", "last_name", "language_code"):
+        if not getattr(target, attr) and getattr(source, attr):
+            setattr(target, attr, getattr(source, attr))
+    if not target.referred_by_id and source.referred_by_id != target_user_id:
+        target.referred_by_id = source.referred_by_id
+    if target.referred_by_id == source_user_id:
+        target.referred_by_id = source.referred_by_id
+        if target.referred_by_id == target_user_id:
+            target.referred_by_id = None
+
+    target_method_ids = select(UserPaymentMethod.provider_payment_method_id).where(
+        UserPaymentMethod.user_id == target_user_id
+    )
+    await session.execute(
+        delete(UserPaymentMethod).where(
+            UserPaymentMethod.user_id == source_user_id,
+            UserPaymentMethod.provider_payment_method_id.in_(target_method_ids),
+        )
+    )
+
+    target_promo_ids = select(PromoCodeActivation.promo_code_id).where(
+        PromoCodeActivation.user_id == target_user_id
+    )
+    await session.execute(
+        delete(PromoCodeActivation).where(
+            PromoCodeActivation.user_id == source_user_id,
+            PromoCodeActivation.promo_code_id.in_(target_promo_ids),
+        )
+    )
+
+    target_has_billing = (
+        await session.execute(
+            select(UserBilling.user_id).where(UserBilling.user_id == target_user_id)
+        )
+    ).scalar_one_or_none()
+    if target_has_billing:
+        await session.execute(delete(UserBilling).where(UserBilling.user_id == source_user_id))
+    else:
+        await session.execute(
+            update(UserBilling)
+            .where(UserBilling.user_id == source_user_id)
+            .values(user_id=target_user_id)
+        )
+
+    target_has_attribution = (
+        await session.execute(
+            select(AdAttribution.user_id).where(AdAttribution.user_id == target_user_id)
+        )
+    ).scalar_one_or_none()
+    if target_has_attribution:
+        await session.execute(
+            delete(AdAttribution).where(AdAttribution.user_id == source_user_id)
+        )
+    else:
+        await session.execute(
+            update(AdAttribution)
+            .where(AdAttribution.user_id == source_user_id)
+            .values(user_id=target_user_id)
+        )
+
+    for model in (Subscription, Payment, PromoCodeActivation, UserPaymentMethod):
+        await session.execute(
+            update(model)
+            .where(model.user_id == source_user_id)
+            .values(user_id=target_user_id)
+        )
+
+    await session.execute(
+        update(MessageLog)
+        .where(MessageLog.user_id == source_user_id)
+        .values(user_id=target_user_id)
+    )
+    await session.execute(
+        update(MessageLog)
+        .where(MessageLog.target_user_id == source_user_id)
+        .values(target_user_id=target_user_id)
+    )
+    await session.execute(
+        update(User)
+        .where(User.referred_by_id == source_user_id)
+        .values(referred_by_id=target_user_id)
+    )
+
+    await session.delete(source)
+    await session.flush()
+    await session.refresh(target)
+    return target
+
+
 async def get_user_by_referral_code(session: AsyncSession, referral_code: str) -> Optional[User]:
     normalized = referral_code.strip().upper()
     if not normalized:

db/migrator.py

@@ -125,6 +125,83 @@ def _migration_0004_add_lifetime_used_traffic(connection: Connection) -> None:
         )
     )
 
+
+def _migration_0005_add_email_auth_fields(connection: Connection) -> None:
+    inspector = inspect(connection)
+    columns: Set[str] = {col["name"] for col in inspector.get_columns("users")}
+
+    if "email" not in columns:
+        connection.execute(text("ALTER TABLE users ADD COLUMN email VARCHAR"))
+    if "email_verified_at" not in columns:
+        connection.execute(
+            text("ALTER TABLE users ADD COLUMN email_verified_at TIMESTAMPTZ")
+        )
+    if "telegram_id" not in columns:
+        connection.execute(text("ALTER TABLE users ADD COLUMN telegram_id BIGINT"))
+
+    connection.execute(
+        text(
+            """
+            UPDATE users
+            SET telegram_id = user_id
+            WHERE telegram_id IS NULL
+              AND user_id > 0
+            """
+        )
+    )
+    connection.execute(
+        text(
+            """
+            CREATE UNIQUE INDEX IF NOT EXISTS uq_users_email
+            ON users (email)
+            WHERE email IS NOT NULL
+            """
+        )
+    )
+    connection.execute(
+        text(
+            """
+            CREATE UNIQUE INDEX IF NOT EXISTS uq_users_telegram_id
+            ON users (telegram_id)
+            WHERE telegram_id IS NOT NULL
+            """
+        )
+    )
+
+    connection.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS email_verification_codes (
+                code_id SERIAL PRIMARY KEY,
+                email VARCHAR NOT NULL,
+                code_hash VARCHAR NOT NULL,
+                purpose VARCHAR NOT NULL,
+                target_user_id BIGINT NULL REFERENCES users(user_id),
+                expires_at TIMESTAMPTZ NOT NULL,
+                consumed_at TIMESTAMPTZ NULL,
+                attempts INTEGER NOT NULL DEFAULT 0,
+                created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+            )
+            """
+        )
+    )
+    connection.execute(
+        text(
+            """
+            CREATE INDEX IF NOT EXISTS ix_email_verification_codes_lookup
+            ON email_verification_codes (email, purpose, target_user_id, created_at DESC)
+            """
+        )
+    )
+    connection.execute(
+        text(
+            """
+            CREATE INDEX IF NOT EXISTS ix_email_verification_codes_expires_at
+            ON email_verification_codes (expires_at)
+            """
+        )
+    )
+
 MIGRATIONS: List[Migration] = [
     Migration(
         id="0001_add_channel_subscription_fields",
@@ -146,6 +223,11 @@ MIGRATIONS: List[Migration] = [
         description="Store lifetime traffic usage for users",
         upgrade=_migration_0004_add_lifetime_used_traffic,
     ),
+    Migration(
+        id="0005_add_email_auth_fields",
+        description="Add email login identities and verification codes",
+        upgrade=_migration_0005_add_email_auth_fields,
+    ),
 ]
 
 

db/models.py

@@ -14,6 +14,9 @@ class User(Base):
 
     user_id = Column(BigInteger, primary_key=True, index=True)
     username = Column(String, nullable=True, index=True)
+    email = Column(String, nullable=True, unique=True, index=True)
+    email_verified_at = Column(DateTime(timezone=True), nullable=True)
+    telegram_id = Column(BigInteger, nullable=True, unique=True, index=True)
     first_name = Column(String, nullable=True)
     last_name = Column(String, nullable=True)
     language_code = Column(String, default="ru")
@@ -86,6 +89,27 @@ class Subscription(Base):
         return f"<Subscription(id={self.subscription_id}, user_id={self.user_id}, panel_uuid='{self.panel_user_uuid}', ends='{self.end_date}')>"
 
 
+class EmailVerificationCode(Base):
+    __tablename__ = "email_verification_codes"
+
+    code_id = Column(Integer, primary_key=True, autoincrement=True)
+    email = Column(String, nullable=False, index=True)
+    code_hash = Column(String, nullable=False)
+    purpose = Column(String, nullable=False, index=True)
+    target_user_id = Column(
+        BigInteger,
+        ForeignKey("users.user_id"),
+        nullable=True,
+        index=True,
+    )
+    expires_at = Column(DateTime(timezone=True), nullable=False, index=True)
+    consumed_at = Column(DateTime(timezone=True), nullable=True)
+    attempts = Column(Integer, nullable=False, default=0)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+    target_user = relationship("User")
+
+
 class Payment(Base):
     __tablename__ = "payments"