fix: replaced remnawave_sdk to local gitea

.gitignore

@@ -180,4 +180,3 @@ plans
 dev.sh
 test_dummy.py
 *.pem
-tests/

alembic/versions/135672cdf14a_user_referal_code.py

@@ -0,0 +1,34 @@
+"""+user.referal_code
+
+Revision ID: 135672cdf14a
+Revises: 72e78a8a43cf
+Create Date: 2026-08-03 10:26:22.290547
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '135672cdf14a'
+down_revision: Union[str, Sequence[str], None] = '72e78a8a43cf'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('users', sa.Column('referal_code', sa.TEXT(), nullable=False))
+    op.create_unique_constraint(None, 'users', ['referal_code'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'users', type_='unique')
+    op.drop_column('users', 'referal_code')
+    # ### end Alembic commands ###

alembic/versions/426b372286c3_add_rw_sync_outbox.py

@@ -0,0 +1,46 @@
+"""add rw sync outbox
+
+Revision ID: 426b372286c3
+Revises: af1c3d7e4b20
+Create Date: 2026-08-20 21:31:20.949847
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '426b372286c3'
+down_revision: Union[str, Sequence[str], None] = 'af1c3d7e4b20'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('rw_sync_outbox',
+    sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('subscription_id', sa.INTEGER(), nullable=False),
+    sa.Column('revision', sa.INTEGER(), nullable=False),
+    sa.Column('attempts', sa.INTEGER(), nullable=False),
+    sa.Column('next_attempt_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('locked_until', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+    sa.ForeignKeyConstraint(['subscription_id'], ['subscriptions.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('subscription_id')
+    )
+    op.create_unique_constraint(None, 'service_notifications', ['id'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'service_notifications', type_='unique')
+    op.drop_table('rw_sync_outbox')
+    # ### end Alembic commands ###

alembic/versions/551c0ad261cd_service_signatures.py

@@ -0,0 +1,41 @@
+"""+service_signatures
+
+Revision ID: 551c0ad261cd
+Revises: cc6625f7dd7f
+Create Date: 2026-08-18 19:17:33.861149
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '551c0ad261cd'
+down_revision: Union[str, Sequence[str], None] = 'cc6625f7dd7f'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('service_signatures',
+    sa.Column('kid', sa.TEXT(), nullable=False),
+    sa.Column('public_key', sa.TEXT(), nullable=False),
+    sa.Column('service_name', sa.TEXT(), nullable=True),
+    sa.Column('status', sa.Enum('ACTIVE', 'INACTIVE', name='servicesignaturestatus'), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('revoked_at', sa.DateTime(timezone=True), nullable=True),
+    sa.PrimaryKeyConstraint('kid'),
+    sa.UniqueConstraint('kid')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('service_signatures')
+    # ### end Alembic commands ###

alembic/versions/6d1bae3dc723_link_codes.py

@@ -0,0 +1,44 @@
+"""+link_codes
+
+Revision ID: 6d1bae3dc723
+Revises: 551c0ad261cd
+Create Date: 2026-08-18 20:19:35.961227
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '6d1bae3dc723'
+down_revision: Union[str, Sequence[str], None] = '551c0ad261cd'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('link_codes',
+    sa.Column('code', sa.TEXT(), nullable=False),
+    sa.Column('user_id', sa.INTEGER(), nullable=False),
+    sa.Column('status', sa.Enum('ACTIVE', 'USED', 'EXPIRED', name='linkcodestatus'), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('expires_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('used_at', sa.DateTime(timezone=True), nullable=True),
+    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
+    sa.PrimaryKeyConstraint('code'),
+    sa.UniqueConstraint('code')
+    )
+    op.create_unique_constraint(None, 'service_signatures', ['kid'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'service_signatures', type_='unique')
+    op.drop_table('link_codes')
+    # ### end Alembic commands ###

alembic/versions/ad537d63c440_enforce_addon_is_enabled_not_null.py

@@ -0,0 +1,38 @@
+"""enforce addon.is_enabled->NOT NULL
+
+Revision ID: ad537d63c440
+Revises: 6d1bae3dc723
+Create Date: 2026-08-19 21:14:39.928563
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'ad537d63c440'
+down_revision: Union[str, Sequence[str], None] = '6d1bae3dc723'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('addons', 'is_enabled',
+               existing_type=sa.BOOLEAN(),
+               nullable=False)
+    op.create_unique_constraint(None, 'link_codes', ['code'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'link_codes', type_='unique')
+    op.alter_column('addons', 'is_enabled',
+               existing_type=sa.BOOLEAN(),
+               nullable=True)
+    # ### end Alembic commands ###

alembic/versions/af1c3d7e4b20_remove_service_notification_expiry_unique.py

@@ -0,0 +1,32 @@
+"""Remove global service notification expiry uniqueness.
+
+Revision ID: af1c3d7e4b20
+Revises: ee6174eeef14
+Create Date: 2026-08-20 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+revision: str = "af1c3d7e4b20"
+down_revision: Union[str, Sequence[str], None] = "ee6174eeef14"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.drop_constraint(
+        "service_notifications_sub_expires_at_key",
+        "service_notifications",
+        type_="unique",
+    )
+
+
+def downgrade() -> None:
+    op.create_unique_constraint(
+        "service_notifications_sub_expires_at_key",
+        "service_notifications",
+        ["sub_expires_at"],
+    )

alembic/versions/cc6625f7dd7f_addon_is_enabled.py

@@ -0,0 +1,32 @@
+"""+addon.is_enabled
+
+Revision ID: cc6625f7dd7f
+Revises: f4ece5936e3d
+Create Date: 2026-08-17 21:39:00.054081
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'cc6625f7dd7f'
+down_revision: Union[str, Sequence[str], None] = 'f4ece5936e3d'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('addons', sa.Column('is_enabled', sa.BOOLEAN(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('addons', 'is_enabled')
+    # ### end Alembic commands ###

alembic/versions/ee6174eeef14_service_notifications.py

@@ -0,0 +1,47 @@
+"""+service_notifications
+
+Revision ID: ee6174eeef14
+Revises: ad537d63c440
+Create Date: 2026-08-20 12:01:51.073160
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'ee6174eeef14'
+down_revision: Union[str, Sequence[str], None] = 'ad537d63c440'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('service_notifications',
+    sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('subscription_id', sa.INTEGER(), nullable=False),
+    sa.Column('notify_type', sa.Enum('SEVEN_DAYS', 'THREE_DAYS', 'ONE_DAY', 'EXPIRED', name='notificationtype'), nullable=False),
+    sa.Column('sub_expires_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('status', sa.Enum('PENDING', 'DISPATCHED', 'SENT', 'FAILED', name='notificationstatus'), nullable=False),
+    sa.Column('dispatched_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('sent_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('attempts', sa.INTEGER(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.ForeignKeyConstraint(['subscription_id'], ['subscriptions.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('id'),
+    sa.UniqueConstraint('sub_expires_at'),
+    sa.UniqueConstraint('subscription_id', 'notify_type', 'sub_expires_at', name='uq_subscription_notification')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('service_notifications')
+    # ### end Alembic commands ###

alembic/versions/f4ece5936e3d_subscription_duration_days.py

@@ -0,0 +1,32 @@
+"""+subscription.duration_days
+
+Revision ID: f4ece5936e3d
+Revises: 135672cdf14a
+Create Date: 2026-08-17 21:23:17.900314
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'f4ece5936e3d'
+down_revision: Union[str, Sequence[str], None] = '135672cdf14a'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('subscriptions', sa.Column('duration_days', sa.INTEGER(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('subscriptions', 'duration_days')
+    # ### end Alembic commands ###

compose.local.yml

@@ -0,0 +1,4 @@
+services:
+  postgres:
+    ports: !override
+      - "5432:5432"

config.py

@@ -5,27 +5,45 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
 class Settings(BaseSettings):
-    model_config = SettingsConfigDict(env_file=".env")
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
 
+    ### Internal settings ###
     postgres_user: str = Field()
     postgres_password: str = Field()
     postgres_host: str = Field()
     postgres_port: str = Field()
     postgres_db: str = Field()
 
+    ### Telegram ###
+    bot_username: str = Field()
+
+    ### Remnawave ###
+    remnawave_base_url: str = Field()
+    remnawave_sub_url: str = Field()
+    remnawave_token: str = Field()
+    remnawave_default_squads_raw: str = Field(alias="REMNAWAVE_DEFAULT_SQUADS_UUIDS")
+
+    ### Finance ###
+    minimal_deposit: int = Field()
+    referal_bonus: int = Field(30)
+
+    #### Pally ####
+    pally_shop_id: str = Field()
+    pally_token: str = Field()
+
+    ### Security related ###
     private_key_fp: str = Field()
     public_key_fp: str = Field()
 
     access_token_ttl: int = Field(description="Access token TTL (minutes)")
+    link_code_ttl: int = Field(8, description="Link Code TTL (minutes)")
+    notification_scan_interval: int = Field(5, ge=1)
+    rw_sync_interval_minutes: int = Field(1, ge=1)
+    rw_sync_reconcile_interval_minutes: int = Field(60, ge=1)
 
-    pally_shop_id: str = Field()
+    min_password_length: int = Field(8)
-    pally_token: str = Field()
+    link_code_length: int = Field(8)
-
+    password_security_threshold: int = Field(2)
-    remnawave_base_url: str = Field()
-    remnawave_token: str = Field()
-    remnawave_default_squads_raw: str = Field(alias="REMNAWAVE_DEFAULT_SQUADS_UUIDS")
-
-    minimal_deposit: int = Field()
 
     @computed_field
     @property
@@ -53,5 +71,10 @@ class Settings(BaseSettings):
         with open(self.public_key_fp, "rb") as f:
             return f.read()
 
+    @computed_field
+    @property
+    def bot_url(self) -> str:
+        return "https://t.me/" + self.bot_username.lstrip("@")
+
 
 cfg = Settings()  # type: ignore

core/auth/fetch_sub.py

@@ -0,0 +1,14 @@
+from db.models import User
+from db.session import UnitOfWork
+from repositories.users import UserRepository
+from schemas.jwt import ServiceJWTPayload
+
+
+async def fetch_subject_from_service(payload: ServiceJWTPayload, uow: UnitOfWork) -> User | None:
+    repo = UserRepository(uow)
+
+    if payload.acting_as.startswith("telegram:"):
+        telegram_id = int(payload.acting_as.split("telegram:")[1])
+        return await repo.get_user_by_telegram_id(telegram_id)
+
+    return

core/auth/jwt.py

@@ -2,26 +2,56 @@ from datetime import UTC, datetime
 
 from fastapi import HTTPException
 from pydantic import ValidationError
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from core.secrets import decode_jwt
+from core.auth.fetch_sub import fetch_subject_from_service
+from core.secrets import decode_jwt, decode_user_jwt, get_kid_from_token
+from db.session import UnitOfWork
+from repositories.service_signatures import get_active_signature_by_kid
 from repositories.users import UserRepository
 from schemas.dto import AuthContext
-from schemas.jwt import JWTPayload
+from schemas.jwt import ServiceJWTPayload, UserJWTPayload
 
 
-async def authorize(token: str, session: AsyncSession, service: str | None = None) -> AuthContext:
+async def authorize_bot(kid: str, token: str, uow: UnitOfWork) -> AuthContext:
-    content = decode_jwt(token)
+    signature = await get_active_signature_by_kid(uow.session, kid)
+
+    if not signature:
+        raise HTTPException(401, detail="Invalid service signature")
+
+    content = decode_jwt(token, signature.public_key, algo="EdDSA")
 
     try:
-        payload = JWTPayload.model_validate(content)
+        payload = ServiceJWTPayload.model_validate(content)
     except ValidationError:
         raise HTTPException(status_code=401, detail="Invalid credentials") from None
 
     if payload.exp < datetime.now(UTC).timestamp():
         raise HTTPException(status_code=401, detail="Access token expired")
 
-    repo = UserRepository(session)
+    subject = await fetch_subject_from_service(payload, uow)
+    if not subject:
+        raise HTTPException(status_code=401, detail="User not found")
+
+    return AuthContext(subject, auth_method="service", service=kid)
+
+
+async def authorize(token: str, uow: UnitOfWork, service: str | None = None) -> AuthContext:
+    kid = get_kid_from_token(token)
+
+    if kid:
+        return await authorize_bot(kid, token, uow)
+
+    content = decode_user_jwt(token)
+
+    try:
+        payload = UserJWTPayload.model_validate(content)
+    except ValidationError:
+        raise HTTPException(status_code=401, detail="Invalid credentials") from None
+
+    if payload.exp < datetime.now(UTC).timestamp():
+        raise HTTPException(status_code=401, detail="Access token expired")
+
+    repo = UserRepository(uow)
     user = await repo.get_user_by_id(int(payload.sub))
 
     if not user:

core/deps.py

@@ -1,16 +1,17 @@
 from fastapi import Depends, HTTPException, Request
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import cfg
 from core.auth import jwt
-from db.session import get_db
+from core.secrets import get_kid_from_token
+from db.session import UnitOfWork, get_uow
 from external.pally import PallyClient
-from schemas.dto import AuthContext
+from repositories.service_signatures import get_active_signature_by_kid
+from schemas.dto import AuthContext, ServiceIdentity
 from services.subscriptions import sync_user_subscription
 
 
 async def get_auth_context(
-    request: Request, session: AsyncSession = Depends(get_db)
+    request: Request, uow: UnitOfWork = Depends(get_uow)
 ) -> AuthContext | None:
     auth = request.headers.get("Authorization")
 
@@ -19,11 +20,29 @@ async def get_auth_context(
 
     if auth.startswith("Bearer"):
         token = auth.removeprefix("Bearer ").strip()
-        ctx = await jwt.authorize(token, session)
+        ctx = await jwt.authorize(token, uow)
-        await sync_user_subscription(session, user=ctx.user)
+        await sync_user_subscription(uow.session, user=ctx.user)
-        await session.commit()
+        await uow.commit()
         return ctx
 
 
+async def get_service_identity(
+    request: Request, uow: UnitOfWork = Depends(get_uow)
+) -> ServiceIdentity | None:
+    auth = request.headers.get("Authorization")
+    if not auth:
+        raise HTTPException(401)
+
+    if auth.startswith("Bearer"):
+        token = auth.removeprefix("Bearer ").strip()
+        kid = get_kid_from_token(token)
+        if not kid:
+            raise HTTPException(401, detail="No kid provided.")
+        signature = await get_active_signature_by_kid(uow.session, kid)
+        if not signature:
+            raise HTTPException(401, detail="Invalid signature")
+        return ServiceIdentity(service=signature.kid)
+
+
 def get_pally_client() -> PallyClient:
     return PallyClient(api_token=cfg.pally_token, payer_pays_commission=True)

core/secrets.py

@@ -1,15 +1,16 @@
 import hashlib
 import logging
 import secrets
-from typing import Any
+from typing import Any, Literal
 
 import jwt
 from argon2 import PasswordHasher
 from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
+from zxcvbn import zxcvbn
 
 from config import cfg
 from schemas.dto import KeyPair
-from schemas.jwt import JWTPayload
+from schemas.jwt import UserJWTPayload
 from schemas.providers import ProvidersType
 
 ctx = PasswordHasher()
@@ -35,15 +36,29 @@ def generate_jwt(payload: dict[str, Any]) -> str:
     return jwt.encode(payload, cfg.private_key, "EdDSA")
 
 
-def decode_jwt(token: str) -> dict[str, Any] | None:
+def get_kid_from_token(token: str) -> str | None:
     try:
-        return jwt.decode(token, cfg.public_key, "EdDSA")
+        header = jwt.get_unverified_header(token)
-    except jwt.ExpiredSignatureError:
+        return header.get("kid")
+    except jwt.exceptions.PyJWTError:
         return
 
 
+def decode_jwt(
+    token: str, public_key: str, algo: Literal["EdDSA"] = "EdDSA"
+) -> dict[str, Any] | None:
+    try:
+        return jwt.decode(token, public_key, algo)
+    except jwt.exceptions.PyJWTError:
+        return
+
+
+def decode_user_jwt(token: str) -> dict[str, Any] | None:
+    return decode_jwt(token, cfg.public_key, algo="EdDSA")
+
+
 def generate_pair(user_id: int, iss: ProvidersType) -> KeyPair:
-    payload = JWTPayload(sub=str(user_id), iss=iss)
+    payload = UserJWTPayload(sub=str(user_id), iss=iss)
     access_token = generate_jwt(payload.model_dump())
     refresh_token = secrets.token_urlsafe(32)
 
@@ -52,3 +67,11 @@ def generate_pair(user_id: int, iss: ProvidersType) -> KeyPair:
 
 def hash_refresh_token(token: str):
     return hashlib.sha256(token.encode()).hexdigest()
+
+
+def estimate_password_strength(password: str) -> bool:
+    if len(password) < cfg.min_password_length:
+        return False
+
+    r = zxcvbn(password)
+    return r.get("score", 0) > cfg.password_security_threshold

db/models/__init__.py

@@ -1,7 +1,11 @@
 from .addons import Addon
 from .invoice import Invoice
+from .link_codes import LinkCode
 from .orders import Order, OrderAddon
 from .pricing import PricingConfig
+from .rw_sync_outbox import RWSyncOutbox
+from .service_notifications import ServiceNotification
+from .service_signatures import ServiceSignature
 from .sessions import Session
 from .subscription_addons import SubscriptionAddon
 from .subscriptions import Subscription
@@ -12,9 +16,13 @@ __all__ = [
     "Addon",
     "BalanceTransaction",
     "Invoice",
+    "LinkCode",
     "Order",
     "OrderAddon",
     "PricingConfig",
+    "RWSyncOutbox",
+    "ServiceNotification",
+    "ServiceSignature",
     "Session",
     "Subscription",
     "SubscriptionAddon",

db/models/addons.py

@@ -1,4 +1,4 @@
-from sqlalchemy import FLOAT, INTEGER, TEXT
+from sqlalchemy import BOOLEAN, FLOAT, INTEGER, TEXT
 from sqlalchemy.orm import Mapped, mapped_column
 
 from db.base import Base
@@ -11,3 +11,4 @@ class Addon(Base):
     name: Mapped[str] = mapped_column(TEXT, nullable=False, unique=False)
     price: Mapped[float] = mapped_column(FLOAT, nullable=False)
     free_threshold: Mapped[int] = mapped_column(INTEGER, default=-1)
+    is_enabled: Mapped[bool] = mapped_column(BOOLEAN, default=False, nullable=False)

db/models/link_codes.py

@@ -0,0 +1,28 @@
+from datetime import UTC, datetime
+
+from sqlalchemy import TEXT, DateTime, Enum, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column
+
+from db.base import Base
+from schemas.enums import LinkCodeStatus
+
+
+class LinkCode(Base):
+    __tablename__ = "link_codes"
+
+    code: Mapped[str] = mapped_column(TEXT, nullable=False, unique=True, primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
+    status: Mapped[LinkCodeStatus] = mapped_column(
+        Enum(LinkCodeStatus, name="linkcodestatus"), nullable=False, default=LinkCodeStatus.ACTIVE
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+    )
+    expires_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+    )
+    used_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)

db/models/rw_sync_outbox.py

@@ -0,0 +1,34 @@
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import INTEGER, DateTime, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from db.base import Base
+
+if TYPE_CHECKING:
+    from db.models import Subscription
+
+
+class RWSyncOutbox(Base):
+    __tablename__ = "rw_sync_outbox"
+
+    id: Mapped[int] = mapped_column(INTEGER, primary_key=True, autoincrement=True)
+    subscription_id: Mapped[int] = mapped_column(
+        ForeignKey("subscriptions.id"), nullable=False, unique=True
+    )
+    revision: Mapped[int] = mapped_column(INTEGER, nullable=False, default=1)
+    attempts: Mapped[int] = mapped_column(INTEGER, nullable=False, default=0)
+    next_attempt_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    locked_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+    )
+
+    subscription: Mapped["Subscription"] = relationship("Subscription", lazy="selectin")

db/models/service_notifications.py

@@ -0,0 +1,52 @@
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import INTEGER, DateTime, Enum, ForeignKey, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from db.base import Base
+from schemas.enums import NotificationStatus, NotificationType
+
+if TYPE_CHECKING:
+    from db.models import Subscription
+
+
+class ServiceNotification(Base):
+    __tablename__ = "service_notifications"
+
+    id: Mapped[int] = mapped_column(
+        INTEGER, unique=True, autoincrement=True, primary_key=True, nullable=False
+    )
+    subscription_id: Mapped[int] = mapped_column(
+        ForeignKey("subscriptions.id"), nullable=False, unique=False
+    )
+    notify_type: Mapped[NotificationType] = mapped_column(
+        Enum(NotificationType, name="notificationtype"), nullable=False
+    )
+    sub_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+    status: Mapped[NotificationStatus] = mapped_column(
+        Enum(NotificationStatus, name="notificationstatus"),
+        nullable=False,
+        default=NotificationStatus.PENDING,
+    )
+    dispatched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)
+    sent_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)
+    attempts: Mapped[int] = mapped_column(INTEGER, nullable=False, default=0)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "subscription_id",
+            "notify_type",
+            "sub_expires_at",
+            name="uq_subscription_notification",
+        ),
+    )
+
+    subscription: Mapped["Subscription"] = relationship("Subscription", lazy="selectin")

db/models/service_signatures.py

@@ -0,0 +1,26 @@
+from datetime import UTC, datetime
+
+from sqlalchemy import TEXT, DateTime, Enum
+from sqlalchemy.orm import Mapped, mapped_column
+
+from db.base import Base
+from schemas.enums import ServiceSignatureStatus
+
+
+class ServiceSignature(Base):
+    __tablename__ = "service_signatures"
+
+    kid: Mapped[str] = mapped_column(TEXT, unique=True, nullable=False, primary_key=True)
+    public_key: Mapped[str] = mapped_column(TEXT, nullable=False)
+    service_name: Mapped[str] = mapped_column(TEXT, nullable=True)
+    status: Mapped[ServiceSignatureStatus] = mapped_column(
+        Enum(ServiceSignatureStatus, name="servicesignaturestatus"),
+        nullable=False,
+        default=ServiceSignatureStatus.INACTIVE,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+    )
+    revoked_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=True)

db/models/subscriptions.py

@@ -20,6 +20,7 @@ class Subscription(Base):
     user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, unique=True)
 
     devices: Mapped[int] = mapped_column(INTEGER, nullable=False)
+    duration_days: Mapped[int] = mapped_column(INTEGER, default=30, nullable=True)
     status: Mapped[SubscriptionStatus] = mapped_column(
         Enum(SubscriptionStatus, name="subscriptionstatus"),
         nullable=False,

db/models/users.py

@@ -1,3 +1,5 @@
+import secrets
+import string
 from typing import TYPE_CHECKING
 
 from sqlalchemy import BIGINT, INTEGER, REAL, TEXT, VARCHAR, ForeignKey
@@ -9,6 +11,11 @@ if TYPE_CHECKING:
     from db.models import Order, Session, Subscription
 
 
+def generate_ref_code(length: int = 8):
+    alphabet = string.ascii_letters + string.digits
+    return "".join(secrets.choice(alphabet) for _ in range(length))
+
+
 class User(Base):
     __tablename__ = "users"
 
@@ -19,6 +26,9 @@ class User(Base):
     hashed_password: Mapped[str] = mapped_column(VARCHAR(255), nullable=True)
     telegram_id: Mapped[int] = mapped_column(BIGINT, unique=True, nullable=True)
     referal_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=True)
+    referal_code: Mapped[str] = mapped_column(
+        TEXT, nullable=False, unique=True, default=generate_ref_code
+    )
 
     balance: Mapped[float] = mapped_column(REAL, nullable=False, default=0)
 
@@ -27,7 +37,4 @@ class User(Base):
         "Subscription", back_populates="user", lazy="selectin"
     )
     sessions: Mapped[list["Session"]] = relationship(back_populates="user", lazy="selectin")
-    referal: Mapped["User | None"] = relationship(
+    referal: Mapped["User | None"] = relationship("User", remote_side=[id], lazy="selectin")
-        "User",
-        remote_side=[id],
-    )

db/session.py

@@ -1,4 +1,4 @@
-from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from config import cfg
 
@@ -6,6 +6,29 @@ engine = create_async_engine(cfg.db_url, echo=True)
 async_session = async_sessionmaker(bind=engine, expire_on_commit=False)
 
 
+class UnitOfWork:
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def commit(self) -> None:
+        await self.session.commit()
+
+    async def rollback(self) -> None:
+        await self.session.rollback()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, *_):
+        if exc_type:
+            await self.rollback()
+
+
 async def get_db():
     async with async_session() as session:
         yield session
+
+
+async def get_uow():
+    async with async_session() as session, UnitOfWork(session) as uow:
+        yield uow

docker-compose.yml

@@ -19,8 +19,6 @@ services:
       interval: 10s
       timeout: 5s
       retries: 5
-    ports:
-      - "5432:5432"
 
 volumes:
   postgres_data:

external/rw.py

@@ -76,10 +76,7 @@ def _parse_user(user_dto) -> RWUserInfo:
     )
 
 
-def _build_rw_username(*, telegram_id: int | None, username: str | None) -> str | None:
+def _build_rw_username(*, username: str | None) -> str | None:
-    if telegram_id is not None:
-        return f"tg_{telegram_id}"
-
     if not username:
         return None
 
@@ -415,7 +412,7 @@ async def delete_hwid(sdk: RemnawaveSDK, user_uuid: str, hwid: str) -> bool:
 
 
 async def update_expire_at(sdk: RemnawaveSDK, user_uuid: str, expire_at: datetime):
-    dto = UpdateUserRequestDto(uuid=user_uuid, expire_at=expire_at)  # type: ignore
+    dto = UpdateUserRequestDto(uuid=UUID(user_uuid), expire_at=expire_at)
     try:
         await sdk.users.update_user(body=dto)
         return True
@@ -431,6 +428,28 @@ def get_sdk() -> RemnawaveSDK | None:
     return RemnawaveSDK(base_url=cfg.remnawave_base_url, token=cfg.remnawave_token)
 
 
+async def get_rw_user(
+    sdk: RemnawaveSDK, telegram_id: int | None = None, username: str | None = None
+) -> RWUserInfo | None:
+    rw_user = None
+    if telegram_id is not None:
+        rw_user = await get_user_by_telegram_id(sdk, telegram_id)
+
+    if rw_user is None:
+        rw_username = _build_rw_username(username=username)
+        if rw_username is None:
+            logger.warning(
+                "Cannot build username for telegram_id=%s username=%s",
+                telegram_id,
+                username,
+            )
+            return None
+
+        rw_user = await get_user_by_username(sdk, rw_username)
+        return rw_user
+    return rw_user
+
+
 async def sync_subscription_by_telegram_id(
     *,
     expires_at: datetime,
@@ -443,38 +462,35 @@ async def sync_subscription_by_telegram_id(
         logger.info("Skipping RW subscription sync: RemnaWave is not configured")
         return False
 
-    rw_user = None
+    rw_user = await get_rw_user(sdk, telegram_id=telegram_id, username=username)
-    if telegram_id is not None:
+    if expires_at <= datetime.now(UTC):
-        rw_user = await get_user_by_telegram_id(sdk, telegram_id)
+        if rw_user is None:
+            return True
+        return await disable_user(sdk, rw_user.uuid)
 
     if rw_user is None:
-        rw_username = _build_rw_username(telegram_id=telegram_id, username=username)
+        rw_username = _build_rw_username(username=username)
-        if rw_username is None:
+        rw_user = await create_user(
-            logger.warning(
+            sdk=sdk,
-                "Skipping RW subscription sync: cannot build username for telegram_id=%s username=%s",
+            username=rw_username,
-                telegram_id,
+            telegram_id=telegram_id,
-                username,
+            expire_at=expires_at,
-            )
+            hwid_device_limit=None if devices == 0 else devices,
-            return False
+            squad_uuids=cfg.remnawave_default_squads,
-
+        )
-        rw_user = await get_user_by_username(sdk, rw_username)
+    if rw_user is None:
-        if rw_user is None:
+        logger.warning(
-            rw_user = await create_user(
+            "Failed to create RW user for telegram_id=%s username=%s",
-                sdk=sdk,
+            telegram_id,
-                username=rw_username,
+            username,
-                telegram_id=telegram_id,
+        )
-                expire_at=expires_at,
+        return False
-                hwid_device_limit=None if devices == 0 else devices,
-                squad_uuids=cfg.remnawave_default_squads,
-            )
-        if rw_user is None:
-            logger.warning(
-                "Failed to create RW user for telegram_id=%s username=%s",
-                telegram_id,
-                username,
-            )
-            return False
 
     expire_synced = await update_expire_at(sdk, rw_user.uuid, expires_at)
     devices_synced = await set_hwid_limit(sdk, rw_user.uuid, devices)
-    return expire_synced and devices_synced
+    enabled = await enable_user(sdk, rw_user.uuid)
+    return expire_synced and devices_synced and enabled
+
+
+def build_subscription_link(short_uuid: str):
+    return cfg.remnawave_sub_url.rstrip("/") + f"/{short_uuid}"

main.py

@@ -1,8 +1,31 @@
+import asyncio
+from contextlib import asynccontextmanager, suppress
+
 from fastapi import FastAPI
 
 from routes import routers
+from services.notifications import run_subscription_notifications
+from services.rw_sync import run_rw_sync_reconciler, run_rw_sync_worker
 
-app = FastAPI(debug=True)
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    tasks = [
+        asyncio.create_task(run_subscription_notifications()),
+        asyncio.create_task(run_rw_sync_worker()),
+        asyncio.create_task(run_rw_sync_reconciler()),
+    ]
+    try:
+        yield
+    finally:
+        for task in tasks:
+            task.cancel()
+        for task in tasks:
+            with suppress(asyncio.CancelledError):
+                await task
+
+
+app = FastAPI(debug=True, lifespan=lifespan)
 
 for r in routers:
     app.include_router(r)

repositories/addons.py

@@ -1,12 +1,12 @@
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from db.models import Addon
+from db.session import UnitOfWork
 
 
 class AddonsRepository:
-    def __init__(self, session: AsyncSession) -> None:
+    def __init__(self, uow: UnitOfWork) -> None:
-        self.session = session
+        self.session = uow.session
 
     async def get_all(self) -> list[Addon]:
         stmt = select(Addon)

repositories/invoices.py

@@ -1,13 +1,14 @@
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from db.models.invoice import Invoice
+from db.session import UnitOfWork
 from schemas.invoices import InvoiceStatus
 
 
 class InvoiceRepository:
-    def __init__(self, session: AsyncSession) -> None:
+    def __init__(self, uow: UnitOfWork) -> None:
-        self.session = session
+        self.uow = uow
+        self.session = uow.session
 
     async def get_by_id(self, id: int) -> Invoice | None:
         stmt = select(Invoice).where(Invoice.id == id)
@@ -32,13 +33,9 @@ class InvoiceRepository:
         )
 
         self.session.add(obj)
-        await self.session.commit()
-
         return obj
 
     async def update_status_by_id(self, invoice_id: int, status: InvoiceStatus) -> Invoice | None:
         invoice = await self.get_by_id(invoice_id)
         invoice.status = status
-        await self.session.commit()
-
         return invoice

repositories/link_codes.py

@@ -0,0 +1,38 @@
+from datetime import datetime
+
+from sqlalchemy import select
+
+from db.models.link_codes import LinkCode
+from db.session import UnitOfWork
+from schemas.enums import LinkCodeStatus
+
+
+async def create_link_code(
+    uow: UnitOfWork,
+    *,
+    code: str,
+    user_id: int,
+    status: LinkCodeStatus,
+    expires_at: datetime,
+) -> LinkCode:
+    link_code = LinkCode(
+        code=code,
+        user_id=user_id,
+        status=status,
+        expires_at=expires_at,
+    )
+
+    uow.session.add(link_code)
+    return link_code
+
+
+async def get_link_code_by_code(uow: UnitOfWork, code: str) -> LinkCode | None:
+    stmt = select(LinkCode).where(LinkCode.code == code)
+    r = await uow.session.execute(stmt)
+
+    return r.scalar_one_or_none()
+
+
+async def use_link_code(uow: UnitOfWork, code: LinkCode) -> LinkCode:
+    code.status = LinkCodeStatus.USED
+    return code

repositories/orders.py

@@ -1,12 +1,13 @@
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from db.models.orders import Order, OrderAddon, OrderStatus
+from db.session import UnitOfWork
 
 
 class OrderRepository:
-    def __init__(self, session: AsyncSession) -> None:
+    def __init__(self, uow: UnitOfWork) -> None:
-        self.session = session
+        self.uow = uow
+        self.session = uow.session
 
     async def create(
         self,
@@ -37,7 +38,7 @@ class OrderRepository:
             addon = OrderAddon(order_id=order.id, addon_id=addon_id)
             self.session.add(addon)
 
-        await self.session.commit()
+        await self.session.refresh(order, attribute_names=["addons"])
 
         return order
 

repositories/pricing.py

@@ -1,12 +1,12 @@
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from db.models.pricing import PricingConfig
+from db.session import UnitOfWork
 
 
 class PricingRepository:
-    def __init__(self, session: AsyncSession) -> None:
+    def __init__(self, uow: UnitOfWork) -> None:
-        self.session = session
+        self.session = uow.session
 
     async def get(self) -> PricingConfig | None:
         stmt = select(PricingConfig).where(PricingConfig.id == 1)

repositories/service_notifications.py

@@ -0,0 +1,61 @@
+from sqlalchemy import func, or_, select, text
+
+from db.models.service_notifications import ServiceNotification
+from db.session import UnitOfWork
+from schemas.enums import NotificationStatus
+
+
+async def get_notification_by_id(uow: UnitOfWork, n_id: int) -> ServiceNotification | None:
+    stmt = select(ServiceNotification).where(ServiceNotification.id == n_id)
+    r = await uow.session.execute(stmt)
+
+    return r.scalar_one_or_none()
+
+
+async def get_pending_notifications(
+    uow: UnitOfWork, batch_size: int = 50
+) -> list[ServiceNotification]:
+    stmt = (
+        select(ServiceNotification)
+        .where(
+            or_(
+                ServiceNotification.status == "pending",
+                (
+                    (ServiceNotification.status == "dispatched")
+                    & (
+                        ServiceNotification.dispatched_at
+                        < func.now() - text("interval '10 minutes'")
+                    )
+                ),
+            )
+        )
+        .order_by(ServiceNotification.created_at)
+        .with_for_update(skip_locked=True)
+        .limit(batch_size)
+    )
+    r = await uow.session.execute(stmt)
+
+    return list(r.scalars().all())
+
+
+async def ack_notification(uow: UnitOfWork, n_id: int) -> ServiceNotification | None:
+    notification = await get_notification_by_id(uow, n_id)
+    if not notification:
+        return
+
+    notification.sent_at = func.now()
+    notification.status = NotificationStatus.SENT
+
+    return notification
+
+
+async def mark_notification_as_dispatched(uow: UnitOfWork, n_id: int):
+    notification = await get_notification_by_id(uow, n_id)
+    if not notification:
+        return
+
+    notification.dispatched_at = func.now()
+    notification.status = NotificationStatus.DISPATCHED
+    notification.attempts += 1
+
+    return notification

repositories/service_signatures.py

@@ -0,0 +1,16 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models.service_signatures import ServiceSignature
+from schemas.enums import ServiceSignatureStatus
+
+
+async def get_active_signature_by_kid(session: AsyncSession, kid: str) -> ServiceSignature | None:
+    stmt = (
+        select(ServiceSignature)
+        .where(ServiceSignature.kid == kid)
+        .where(ServiceSignature.status == ServiceSignatureStatus.ACTIVE)
+    )
+    r = await session.execute(stmt)
+
+    return r.scalar_one_or_none()

repositories/sessions.py

@@ -1,13 +1,14 @@
 from sqlalchemy import func, select
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from db.models import Session
+from db.session import UnitOfWork
 from schemas.providers import ProvidersType
 
 
 class SessionsRepository:
-    def __init__(self, session: AsyncSession) -> None:
+    def __init__(self, uow: UnitOfWork) -> None:
-        self.session = session
+        self.uow = uow
+        self.session = uow.session
 
     async def get_session_by_id(self, id: int) -> Session | None:
         stmt = select(Session).where(Session.id == id)
@@ -27,12 +28,10 @@ class SessionsRepository:
     async def create(self, user_id: int, refresh_token_hash: str, iss: ProvidersType) -> Session:
         obj = Session(user_id=user_id, refresh_token_hash=refresh_token_hash, source=iss)
         self.session.add(obj)
-        await self.session.commit()
         return obj
 
     async def revoke(self, token_id: int):
         session = await self.get_session_by_id(token_id)
         session.is_revoked = True
         session.revoked_at = func.now()
-        await self.session.commit()
         return session

repositories/users.py

@@ -1,13 +1,14 @@
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from db.models import User
 from db.models.transactions import BalanceTransaction, BalanceTxType
+from db.session import UnitOfWork
 
 
 class UserRepository:
-    def __init__(self, session: AsyncSession) -> None:
+    def __init__(self, uow: UnitOfWork) -> None:
-        self.session = session
+        self.uow = uow
+        self.session = uow.session
 
     async def get_user_by_id(self, id: int) -> User | None:
         stmt = select(User).where(User.id == id)
@@ -24,21 +25,26 @@ class UserRepository:
         res = await self.session.execute(stmt)
         return res.scalar_one_or_none()
 
+    async def get_user_by_ref_code(self, ref_code: str) -> User | None:
+        stmt = select(User).where(User.referal_code == ref_code)
+        res = await self.session.execute(stmt)
+        return res.scalar_one_or_none()
+
     async def create(
         self,
         *,
         username: str | None = None,
         hashed_password: str | None = None,
         telegram_id: int | None = None,
+        referal_id: int | None = None,
     ) -> User:
         obj = User(
             username=username,
             hashed_password=hashed_password,
             telegram_id=telegram_id,
+            referal_id=referal_id,
         )
         self.session.add(obj)
-        await self.session.commit()
-
         return obj
 
     async def increase_balance(
@@ -60,5 +66,8 @@ class UserRepository:
         self.session.add(obj)
 
         user.balance += amount
-        await self.session.commit()
+        return user
+
+    async def update_telegram_id(self, user: User, telegram_id: int) -> User:
+        user.telegram_id = telegram_id
         return user

requirements.txt

@@ -8,4 +8,10 @@ asyncpg>=0.31.0
 alembic>=1.18.0
 aiohttp>=3.14.0
 python-multipart==0.0.32
-remnawave>=2.6.1
+# Private remnawave SDK: bare URL without creds (safe to commit).
+# Local install: export REMNAWAVE_SDK_TOKEN=xxx && git config --global url."https://agony:${REMNAWAVE_SDK_TOKEN}@git.mdevs.lat/".insteadOf "https://git.mdevs.lat/" && pip install -r requirements.txt
+# Without the token git clone fails with 401/403 (repo is private, login is hardcoded to 'agony').
+remnawave @ git+https://git.mdevs.lat/agony/remnawave-sdk.git
+zxcvbn>=4.5.0
+httpx2>=2.11.0
+pytest>=9.1.0

routes/__init__.py

@@ -1,8 +1,21 @@
 from fastapi import APIRouter
 
 from .auth import router as auth_router
+from .health import router as health_router
+from .internal import internal_router
+from .link_codes import router as link_code_router
 from .orders import router as orders_router
-from .payments import payment_routers
+from .payments import payment_router
 from .plans import router as plans_router
+from .users import router as users_routers
 
-routers: list[APIRouter] = [auth_router, plans_router, orders_router, *payment_routers]
+routers: list[APIRouter] = [
+    auth_router,
+    plans_router,
+    orders_router,
+    users_routers,
+    health_router,
+    link_code_router,
+    payment_router,
+    internal_router,
+]

routes/auth.py

@@ -1,12 +1,15 @@
 from fastapi import APIRouter, Depends, HTTPException
-from fastapi.responses import JSONResponse
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from core.secrets import hash_password, hash_refresh_token, verify_password
+from core.secrets import (
-from db.session import get_db
+    estimate_password_strength,
+    hash_password,
+    hash_refresh_token,
+    verify_password,
+)
+from db.session import UnitOfWork, get_uow
 from repositories.sessions import SessionsRepository
 from repositories.users import UserRepository
-from schemas.login import UserLogin, UserLoginData, UserTokens
+from schemas.login import AuthenticatedUser, UserLoginData, UserTokens
 from schemas.providers import ProvidersType
 from schemas.registration import UserRegistration
 from schemas.user import UserInfo
@@ -16,9 +19,10 @@ from services.users import authorize_user
 router = APIRouter(prefix="/auth")
 
 
-@router.post("/signup")
+@router.post("/signup", response_model=AuthenticatedUser)
-async def signup(req: UserRegistration, session: AsyncSession = Depends(get_db)):
+async def signup(req: UserRegistration, uow: UnitOfWork = Depends(get_uow)) -> AuthenticatedUser:
-    users_repo = UserRepository(session)
+    users_repo = UserRepository(uow)
+    sessions_repo = SessionsRepository(uow)
 
     if req.provider == "credentials":
         if not req.username or not req.password:
@@ -27,19 +31,40 @@ async def signup(req: UserRegistration, session: AsyncSession = Depends(get_db))
         if user:
             raise HTTPException(status_code=409, detail="User already exists")
 
+        if not estimate_password_strength(req.password):
+            raise HTTPException(422, detail="Password is not secure.")
+
         password_hash = hash_password(req.password)
-        user = await users_repo.create(username=req.username, hashed_password=password_hash)
+        referal_id = None
-        return JSONResponse(
+        if req.referal_code:
-            UserInfo(username=user.username, telegram_id=user.telegram_id).model_dump(),
+            referal = await users_repo.get_user_by_ref_code(req.referal_code)
-            status_code=201,
+            referal_id = referal.id if referal else None
+
+        user = await users_repo.create(
+            username=req.username, hashed_password=password_hash, referal_id=referal_id
+        )
+        await uow.commit()
+
+        info = UserInfo(
+            username=user.username,
+            telegram_id=user.telegram_id,
+            referal_code=user.referal_code,
+            bonus_balance=user.balance,
+        )
+        user_login = await authorize_user(sessions_repo, user, req.provider)
+        return AuthenticatedUser(
+            access_token=user_login.access_token,
+            refresh_token=user_login.refresh_token,
+            expires_at=user_login.expires_at,
+            user=info,
         )
     raise HTTPException(status_code=400, detail="Unsupported provider")
 
 
-@router.post("/login", response_model=UserLogin)
+@router.post("/login", response_model=AuthenticatedUser)
-async def login(req: UserLoginData, session: AsyncSession = Depends(get_db)):
+async def login(req: UserLoginData, uow: UnitOfWork = Depends(get_uow)):
-    users_repo = UserRepository(session)
+    users_repo = UserRepository(uow)
-    sessions_repo = SessionsRepository(session)
+    sessions_repo = SessionsRepository(uow)
     if req.provider == "credentials":
         if not req.username or not req.password:
             raise HTTPException(status_code=400, detail="Username or password is not provided.")
@@ -52,6 +77,7 @@ async def login(req: UserLoginData, session: AsyncSession = Depends(get_db)):
             raise HTTPException(status_code=401, detail="Invalid password")
 
         data = await authorize_user(sessions_repo, user, req.provider)
+        await uow.commit()
         return data
 
     if req.provider == "telegram":
@@ -62,8 +88,8 @@ async def login(req: UserLoginData, session: AsyncSession = Depends(get_db)):
 
 
 @router.post("/refresh", response_model=UserTokens)
-async def refresh(refresh_token: str, iss: ProvidersType, session: AsyncSession = Depends(get_db)):
+async def refresh(refresh_token: str, iss: ProvidersType, uow: UnitOfWork = Depends(get_uow)):
-    sessions_repo = SessionsRepository(session)
+    sessions_repo = SessionsRepository(uow)
 
     token_hash = hash_refresh_token(refresh_token)
     token_entry = await sessions_repo.get_session_by_hash(token_hash)
@@ -72,6 +98,7 @@ async def refresh(refresh_token: str, iss: ProvidersType, session: AsyncSession
         raise HTTPException(status_code=401, detail="Refresh token is invalid.")
 
     key_pair = await refresh_token_rotation(sessions_repo, token_entry, iss)
+    await uow.commit()
     return UserTokens(
         access_token=key_pair.access_token,
         refresh_token=key_pair.refresh_token,

routes/health.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+router = APIRouter(prefix="/health")
+
+
+@router.get("/")
+async def healthcheck():
+    return "OK"

routes/internal/__init__.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+from .renewal import router as renewal_router
+
+internal_router = APIRouter(prefix="/internal")
+internal_router.include_router(renewal_router)

routes/internal/renewal.py

@@ -0,0 +1,60 @@
+from fastapi import APIRouter, Depends, HTTPException
+
+from core.deps import get_service_identity
+from db.session import UnitOfWork, get_uow
+from repositories.service_notifications import (
+    ack_notification,
+    get_pending_notifications,
+    mark_notification_as_dispatched,
+)
+from schemas.dto import ServiceIdentity
+from schemas.notifications import (
+    NotificationAcknowledgeRequest,
+    NotificationResponse,
+    UserNotificationData,
+)
+
+router = APIRouter(prefix="/renewal")
+
+
+@router.get("/pending", response_model=NotificationResponse)
+async def get_pending(
+    limit: int = 50,
+    ctx: ServiceIdentity = Depends(get_service_identity),
+    uow: UnitOfWork = Depends(get_uow),
+):
+    notifications = await get_pending_notifications(uow, limit)
+
+    response_users: list[UserNotificationData] = []
+    for notification in notifications:
+        response_users.append(
+            UserNotificationData(
+                notification_id=notification.id,
+                username=notification.subscription.user.username,
+                telegram_id=notification.subscription.user.telegram_id,
+                expires_at=notification.sub_expires_at,
+                notification_type=notification.notify_type,
+            )
+        )
+
+        await mark_notification_as_dispatched(uow, notification.id)
+
+    await uow.commit()
+    return NotificationResponse(users=response_users, issued_by=ctx.service)
+
+
+@router.post("/ack")
+async def acknowledge(
+    req: NotificationAcknowledgeRequest,
+    ctx: ServiceIdentity = Depends(get_service_identity),
+    uow: UnitOfWork = Depends(get_uow),
+):
+    try:
+        r = await ack_notification(uow, req.notification_id)
+    except Exception as e:
+        raise HTTPException(500, detail=str(e)) from None
+
+    if r:
+        await uow.commit()
+        return "OK"
+    raise HTTPException(500, detail="No such notification found.")

routes/link_codes.py

@@ -0,0 +1,63 @@
+import secrets
+from datetime import UTC, datetime, timedelta
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from config import cfg
+from core.deps import get_auth_context, get_service_identity
+from db.session import UnitOfWork, get_uow
+from repositories.link_codes import create_link_code, get_link_code_by_code, use_link_code
+from repositories.users import UserRepository
+from schemas.dto import AuthContext
+from schemas.enums import LinkCodeStatus
+from schemas.link_codes import LinkCodeConsume, LinkCodeResponse
+from schemas.user import UserInfo
+
+router = APIRouter(prefix="/link-codes")
+
+
+@router.post("", response_model=LinkCodeResponse, status_code=201)
+async def gen_link_code(
+    ctx: AuthContext = Depends(get_auth_context), uow: UnitOfWork = Depends(get_uow)
+):
+    code = secrets.token_urlsafe(cfg.link_code_length)
+    exp = datetime.now(UTC) + timedelta(minutes=cfg.link_code_ttl)
+    link_code = await create_link_code(
+        uow, code=code, user_id=ctx.user.id, status=LinkCodeStatus.ACTIVE, expires_at=exp
+    )
+    await uow.commit()
+
+    return LinkCodeResponse(code=link_code.code, expires_at=link_code.expires_at)
+
+
+@router.post("/consume", response_model=UserInfo)
+async def consume_link_code(
+    payload: LinkCodeConsume,
+    ctx: AuthContext = Depends(get_service_identity),
+    uow: UnitOfWork = Depends(get_uow),
+):
+    link_code = await get_link_code_by_code(uow, payload.code)
+
+    if not link_code:
+        raise HTTPException(404, detail="Code not found")
+
+    if link_code.status != LinkCodeStatus.ACTIVE:
+        raise HTTPException(404, detail="Code expired or is invalid.")
+
+    users_repo = UserRepository(uow)
+    user = await users_repo.get_user_by_id(link_code.user_id)
+
+    if not user:
+        raise HTTPException(404, detail="User not found")
+
+    user = await users_repo.update_telegram_id(user, payload.telegram_id)
+
+    await use_link_code(uow, code=link_code)
+    await uow.commit()
+
+    return UserInfo(
+        username=user.username,
+        telegram_id=user.telegram_id,
+        referal_code=user.referal_code,
+        bonus_balance=user.balance,
+    )

routes/orders.py

@@ -2,12 +2,11 @@ import math
 from datetime import UTC, datetime
 
 from fastapi import APIRouter, Depends, HTTPException
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import cfg
 from core.deps import get_auth_context, get_pally_client
 from db.models.orders import OrderStatus
-from db.session import get_db
+from db.session import UnitOfWork, get_uow
 from external.pally import PallyClient
 from repositories import AddonsRepository, PricingRepository
 from repositories.invoices import InvoiceRepository
@@ -31,13 +30,13 @@ router = APIRouter(prefix="/orders")
 async def checkout(
     order: OrderDetails,
     ctx: AuthContext = Depends(get_auth_context),
-    session: AsyncSession = Depends(get_db),
+    uow: UnitOfWork = Depends(get_uow),
     pally: PallyClient = Depends(get_pally_client),
 ):
-    addons_repo = AddonsRepository(session)
+    addons_repo = AddonsRepository(uow)
-    pricing_repo = PricingRepository(session)
+    pricing_repo = PricingRepository(uow)
-    invoices_repo = InvoiceRepository(session)
+    invoices_repo = InvoiceRepository(uow)
-    orders_repo = OrderRepository(session)
+    orders_repo = OrderRepository(uow)
 
     pricing = await get_pricing_model(addons_repo, pricing_repo)
     price = math.ceil(await calculate_price(addons_repo=addons_repo, order=order, pricing=pricing))
@@ -72,7 +71,7 @@ async def checkout(
             raise HTTPException(500, detail="failed to create invoice")
     else:
         await deduct_order_balance(
-            session,
+            uow.session,
             user=ctx.user,
             order=order_entry,
             description=f"order {order_entry.id} paid from balance",
@@ -85,7 +84,7 @@ async def checkout(
             now=now,
         ):
             await apply_order_now(
-                session,
+                uow.session,
                 user=ctx.user,
                 order=order_entry,
                 pricing=pricing,
@@ -95,9 +94,12 @@ async def checkout(
             await queue_order_for_later(
                 order=order_entry, subscription=ctx.user.subscription, now=now
             )
-        await session.commit()
+        await uow.commit()
         payment_link = None
 
+    if amount_to_pay > 0:
+        await uow.commit()
+
     return CheckoutResponse(
         order_id=str(order_entry.id),
         total_amount=price,

routes/payments/__init__.py

@@ -1,3 +1,6 @@
+from fastapi import APIRouter
+
 from .pally import router as pally_router
 
-payment_routers = [pally_router]
+payment_router = APIRouter(prefix="/payments")
+payment_router.include_router(pally_router)

routes/payments/pally.py

@@ -1,41 +1,24 @@
 # ruff: noqa: N803
-import hashlib
-import hmac
 import logging
-import math
-from datetime import UTC, datetime
 
 from fastapi import Depends, Form, HTTPException
 from fastapi.routing import APIRouter
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import cfg
+from db.models.transactions import BalanceTxType
-from core.deps import get_db
+from db.session import UnitOfWork, get_uow
-from db.models.orders import OrderStatus
-from db.models.transactions import BalanceTransaction, BalanceTxType
 from external.pally import BillStatus
-from external.rw import sync_subscription_by_telegram_id
-from repositories import AddonsRepository
 from repositories.invoices import InvoiceRepository
-from repositories.orders import OrderRepository
-from repositories.pricing import PricingRepository
 from repositories.users import UserRepository
 from schemas.invoices import InvoiceStatus
-from services.plans import get_pricing_model
+from services.payments import process_subscription_purchase, validate_pally_signature
-from services.subscriptions import (
-    apply_order_now,
-    deduct_order_balance,
-    queue_order_for_later,
-    should_apply_immediately,
-)
 
-router = APIRouter(prefix="/payments/pally")
+router = APIRouter(prefix="/pally")
 
 logger = logging.getLogger(__name__)
 
 
 @router.post("/result")
-async def pally_callback(  # noqa: PLR0911, PLR0912, PLR0915
+async def pally_callback(
     *,
     InvId: str = Form(...),
     OutSum: str = Form(...),
@@ -45,7 +28,6 @@ async def pally_callback(  # noqa: PLR0911, PLR0912, PLR0915
     CurrencyIn: str = Form(...),
     custom: str | None = Form(None),
     SignatureValue: str = Form(...),
-    # Optional fields for additional information
     AccountType: str | None = Form(None),
     AccountNumber: str | None = Form(None),
     BalanceAmount: str | None = Form(None),
@@ -56,16 +38,8 @@ async def pally_callback(  # noqa: PLR0911, PLR0912, PLR0915
     PayerComment: str | None = Form(None),
     ErrorCode: int | None = Form(None),
     ErrorMessage: str | None = Form(None),
-    session: AsyncSession = Depends(get_db),
+    uow: UnitOfWork = Depends(get_uow),
 ):
-    invoice_repo = InvoiceRepository(session)
-    orders_repo = OrderRepository(session)
-    users_repo = UserRepository(session)
-    pricing_repo = PricingRepository(session)
-    invoice_id_str = InvId
-    invoice_id: int | None = None
-    invoice_creator_id: int | None = None
-
     logger.info(
         "Pally webhook received - InvId: %s, OutSum: %s, Commission: %s, TrsId: %s, Status: %s, "
         "CurrencyIn: %s, custom: %s, BalanceAmount: %s, SignatureValue: %s",
@@ -89,29 +63,18 @@ async def pally_callback(  # noqa: PLR0911, PLR0912, PLR0915
             TrsId,
         )
 
-    # Validate signature
+    if not validate_pally_signature(OutSum, InvId, SignatureValue):
-    raw_string = f"{OutSum}:{InvId}:{cfg.pally_token}"
-    expected_signature = hashlib.md5(raw_string.encode("utf-8")).hexdigest().upper()
-
-    logger.debug("Signature validation for TrsId %s", TrsId)
-
-    if not hmac.compare_digest(SignatureValue, expected_signature):
         logger.critical(
-            "SECURITY ALERT: Invalid signature for TrsId %s - Expected: %s, Received: %s",
+            "SECURITY ALERT: Invalid signature for TrsId %s",
             TrsId,
-            expected_signature,
-            SignatureValue,
         )
         raise HTTPException(403, detail="Invalid signature.")
 
-    # Only process successful payments
     if Status != BillStatus.SUCCESS:
         logger.info("Bill %s skipped: status=%s", TrsId, Status)
         return "OK"
 
-    logger.info("Processing successfully paid bill %s", TrsId)
+    invoice_id_str = InvId
-
-    # Validate invoice ID
     if not invoice_id_str or not invoice_id_str.isdigit():
         logger.critical(
             "Invalid or non-numeric bill ID in InvId field for TrsId %s: '%s'",
@@ -120,199 +83,64 @@ async def pally_callback(  # noqa: PLR0911, PLR0912, PLR0915
         )
         return "OK"
 
-    # Find bill in database
+    amount = int(float(BalanceAmount)) if BalanceAmount is not None else int(float(OutSum))
-    invoice = await invoice_repo.get_by_id(int(invoice_id_str))
-
-    if not invoice:
-        logger.critical("Bill %s not found in database for TrsId %s", invoice_id_str, TrsId)
-        return "OK"
-
-    invoice_id = invoice.id
-    invoice_creator_id = invoice.creator_id
-
-    # Check if already processed
-    if invoice.status != InvoiceStatus.ACTIVE:
-        logger.warning(
-            "Bill %s (TrsId: %s) is already processed with status: %s",
-            invoice.id,
-            TrsId,
-            invoice.status,
-        )
-        return "OK"
 
     try:
-        # Handle fee scenarios: use BalanceAmount if available (net amount after fees),
+        await process_subscription_purchase(
-        # otherwise use OutSum (gross amount paid by customer)
+            uow,
-        if BalanceAmount is not None:
+            invoice_id=int(invoice_id_str),
-            # Customer pays fees - BalanceAmount is the net amount credited to merchant
+            trs_id=TrsId,
-            credited_amount = int(float(BalanceAmount))
+            amount=amount,
-            gross_amount = int(float(OutSum))
+        )
+    except Exception:
+        # The payment is confirmed, so never leave the user without the money
+        # if order/subscription provisioning fails.
+        logger.exception(
+            "Subscription provisioning failed for bill %s (TrsId: %s); "
+            "crediting the user's bonus balance",
+            invoice_id_str,
+            TrsId,
+        )
+        await uow.rollback()
 
-            logger.info(
+        invoice_repo = InvoiceRepository(uow)
-                "Customer-pays-fees payment: bill_id=%s, expected=%s, gross_paid=%s, net_credited=%s",
+        invoice = await invoice_repo.get_by_id(int(invoice_id_str))
-                invoice.id,
+        if invoice is None:
-                invoice.amount,
-                gross_amount,
-                credited_amount,
-            )
-
-            # Validate that the net credited amount matches our bill amount
-            if invoice.amount != credited_amount:
-                logger.error(
-                    "Net amount mismatch for bill %s (TrsId: %s) - Expected: %s, Net credited: %s, Gross paid: %s",
-                    invoice.id,
-                    TrsId,
-                    invoice.amount,
-                    credited_amount,
-                    gross_amount,
-                )
-                return "OK"
-
-            amount = credited_amount  # Credit the net amount (without fees)
-
-        else:
-            # Standard payment - OutSum should match bill amount exactly
-            amount = int(float(OutSum))
-
-            logger.info(
-                "Standard payment: bill_id=%s, expected=%s, received=%s",
-                invoice.id,
-                invoice.amount,
-                amount,
-            )
-
-            if invoice.amount != amount:
-                logger.error(
-                    "Amount mismatch for bill %s (TrsId: %s) - Expected: %s, Received: %s",
-                    invoice.id,
-                    TrsId,
-                    invoice.amount,
-                    amount,
-                )
-                return "OK"
-
-        if invoice.order_id is None:
-            logger.critical("Invoice %s has no linked order for TrsId %s", invoice.id, TrsId)
-            return "OK"
-
-        order = await orders_repo.get_by_id(invoice.order_id)
-        if order is None:
             logger.critical(
-                "Order not found for bill %s (TrsId: %s, user_id=%s, order_id=%s)",
+                "Cannot credit fallback balance: bill %s was not found (TrsId: %s)",
-                invoice.id,
+                invoice_id_str,
-                TrsId,
-                invoice.creator_id,
-                invoice.order_id,
-            )
-            return "OK"
-
-        if order.user_id != invoice.creator_id:
-            logger.critical(
-                "Order %s does not belong to invoice creator %s for TrsId %s",
-                order.id,
-                invoice.creator_id,
                 TrsId,
             )
+            raise
+
+        if invoice.status == InvoiceStatus.PAID:
             return "OK"
 
+        users_repo = UserRepository(uow)
         user = await users_repo.get_user_by_id(invoice.creator_id)
         if user is None:
-            logger.critical("User %s not found for TrsId %s", invoice.creator_id, TrsId)
+            logger.critical(
-            return "OK"
+                "Cannot credit fallback balance: user %s was not found " "for bill %s (TrsId: %s)",
+                invoice.creator_id,
+                invoice_id_str,
+                TrsId,
+            )
+            raise
 
-        subscription = user.subscription
+        await users_repo.increase_balance(
-        now = datetime.now(UTC)
+            user.id,
-        pricing = await get_pricing_model(AddonsRepository(session), pricing_repo)
-
-        logger.info(
-            "Processing payment: bill_id=%s, order_id=%s, user_id=%s, amount=%s",
-            invoice.id,
-            order.id,
-            invoice.creator_id,
             amount,
+            BalanceTxType.DEPOSIT,
+            f"fallback payment credit for invoice {invoice.id} (TrsId: {TrsId})",
         )
-
+        await invoice_repo.update_status_by_id(invoice.id, InvoiceStatus.PAID)
-        order.status = OrderStatus.PAID
+        await uow.commit()
-        invoice.status = InvoiceStatus.PAID
+        logger.info(
-
+            "Fallback payment credit processed: user_id=%s, amount=%s, " "invoice_id=%s, TrsId=%s",
-        if order.balance_amount > 0:
+            user.id,
-            await deduct_order_balance(
+            amount,
-                session,
+            invoice.id,
-                user=user,
-                order=order,
-                description=f"order {order.id} partial payment from balance",
-            )
-
-        if should_apply_immediately(
-            subscription=subscription,
-            order=order,
-            pricing=pricing,
-            now=now,
-        ):
-            applied_subscription = await apply_order_now(
-                session, user=user, order=order, pricing=pricing, now=now
-            )
-            await sync_subscription_by_telegram_id(
-                expires_at=applied_subscription.expires_at,
-                devices=applied_subscription.devices,
-                telegram_id=user.telegram_id,
-                username=user.username,
-            )
-        else:
-            if subscription is None:
-                logger.critical(
-                    "Cannot queue order %s without subscription for user %s", order.id, user.id
-                )
-                return "OK"
-            await queue_order_for_later(order=order, subscription=subscription, now=now)
-
-        # Process referral bonus
-        referal_id = user.referal_id
-        if referal_id is not None:
-            referal_amount = math.floor(amount * (cfg.referal_bonus / 100))
-            if referal_amount > 0:
-                referal_user = await session.get(type(user), referal_id)
-                if referal_user is not None:
-                    session.add(
-                        BalanceTransaction(
-                            user_id=referal_id,
-                            amount=referal_amount,
-                            tx_type=BalanceTxType.REFERRAL_BONUS,
-                            balance_before=referal_user.balance,
-                            balance_after=referal_user.balance + referal_amount,
-                            description=f"referral reward for user {invoice.creator_id} (TrsId: {TrsId})",
-                        )
-                    )
-                    referal_user.balance += referal_amount
-
-                    logger.info(
-                        "Referral bonus processed: referrer_id=%s, amount=%s",
-                        referal_id,
-                        referal_amount,
-                    )
-                else:
-                    logger.warning(
-                        "Referrer %s not found for user_id=%s while processing TrsId %s",
-                        referal_id,
-                        invoice.creator_id,
-                        TrsId,
-                    )
-
-        await session.commit()
-
-        logger.info("Payment processing completed successfully for TrsId %s", TrsId)
-
-    except Exception as e:
-        await session.rollback()
-        logger.exception(
-            "CRITICAL ERROR processing payment for TrsId %s, bill_id %s, user_id %s: %s",
             TrsId,
-            invoice_id,
-            invoice_creator_id,
-            str(e),
         )
-        raise HTTPException(500, detail="Payment processing failed.") from e
-
-    logger.info("Bill %s marked as PAID for TrsId %s", invoice_id, TrsId)
 
     return "OK"

routes/plans.py

@@ -1,7 +1,6 @@
 from fastapi import APIRouter, Depends
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from db.session import get_db
+from db.session import UnitOfWork, get_uow
 from repositories.addons import AddonsRepository
 from repositories.pricing import PricingRepository
 from schemas.plans import PricingPlans
@@ -11,9 +10,9 @@ router = APIRouter(prefix="/plans")
 
 
 @router.get("/", response_model=PricingPlans)
-async def get_plans(session: AsyncSession = Depends(get_db)):
+async def get_plans(uow: UnitOfWork = Depends(get_uow)):
-    addons_repo = AddonsRepository(session)
+    addons_repo = AddonsRepository(uow)
-    pricing_repo = PricingRepository(session)
+    pricing_repo = PricingRepository(uow)
 
     res = await get_pricing_model(addons_repo, pricing_repo)
     return res

routes/users.py

@@ -0,0 +1,97 @@
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from core.deps import get_auth_context
+from external.rw import build_subscription_link, delete_hwid, get_hwid_list, get_rw_user, get_sdk
+from schemas.common import OperationData
+from schemas.devices import Device
+from schemas.dto import AuthContext
+from schemas.user import SubscriptionData, UserInfo
+
+router = APIRouter(prefix="/users")
+logger = logging.getLogger(__name__)
+
+
+@router.get("/me", response_model=UserInfo)
+async def get_me(ctx: AuthContext = Depends(get_auth_context)):
+    return UserInfo(
+        username=ctx.user.username,
+        telegram_id=ctx.user.telegram_id,
+        referal_code=ctx.user.referal_code,
+        bonus_balance=ctx.user.balance,
+    )
+
+
+@router.get("/subscription")
+async def get_subscription(ctx: AuthContext = Depends(get_auth_context)):
+    sub = ctx.user.subscription
+    if not sub:
+        return SubscriptionData(
+            has_subscription=False,
+            devices=None,
+            expires_at=None,
+            addon_ids=[],
+            subscription_link=None,
+            duration_days=None,
+        )
+    rw_user = await get_rw_user(get_sdk(), ctx.user.telegram_id, ctx.user.username)
+
+    if not rw_user:
+        logger.critical(
+            "RW user not found for existing local subscription (user_id=%d, sub_id=%d)",
+            ctx.user.id,
+            sub.id,
+        )
+        return SubscriptionData(
+            has_subscription=True,
+            devices=sub.devices,
+            expires_at=sub.expires_at,
+            addon_ids=[a.addon_id for a in sub.addons],
+            subscription_link=None,
+            duration_days=sub.duration_days,
+        )
+
+    return SubscriptionData(
+        has_subscription=True,
+        devices=sub.devices,
+        expires_at=sub.expires_at,
+        addon_ids=[a.addon_id for a in sub.addons],
+        subscription_link=build_subscription_link(rw_user.short_uuid),
+        duration_days=sub.duration_days,
+    )
+
+
+@router.get("/subscription/hwid", response_model=list[Device])
+async def get_hwid(ctx: AuthContext = Depends(get_auth_context)):
+    rw_user = await get_rw_user(get_sdk(), ctx.user.telegram_id, ctx.user.username)
+    if not rw_user:
+        raise HTTPException(403, detail="RW user not found.")
+    hwid_list = await get_hwid_list(get_sdk(), user_uuid=rw_user.uuid)
+    devices = []
+
+    if not hwid_list:
+        return devices
+
+    for device in hwid_list:
+        devices.append(
+            Device(
+                os=device.platform,
+                model=device.device_model,
+                client=device.user_agent.split("/")[0],
+                hwid=device.hwid,
+            )
+        )
+
+    return devices
+
+
+@router.delete("/subscription/hwid")
+async def delete_hwid_endpoint(hwid: str, ctx: AuthContext = Depends(get_auth_context)):
+    rw_user = await get_rw_user(get_sdk(), ctx.user.telegram_id, ctx.user.username)
+    if not rw_user:
+        raise HTTPException(403, detail="RW user not found.")
+
+    success = await delete_hwid(get_sdk(), rw_user.uuid, hwid)
+
+    return OperationData(success=success)

schemas/common.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel, Field
+
+
+class OperationData(BaseModel):
+    success: bool = Field(False)

schemas/devices.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel, Field
+
+
+class Device(BaseModel):
+    os: str | None = Field(None)
+    model: str | None = Field(None)
+    client: str | None = Field(None)
+    hwid: str = Field()

schemas/dto.py

@@ -14,5 +14,14 @@ class KeyPair:
 @dataclass
 class AuthContext:
     user: User
-    auth_method: Literal["jwt"]
+    auth_method: Literal["jwt", "service"]
     service: str | None = None
+
+
+@dataclass
+class ServiceIdentity:
+    """
+    Proves that request is coming from verified source, no user context provided.
+    """
+
+    service: str

schemas/enums.py

@@ -4,3 +4,28 @@ from enum import StrEnum
 class SubscriptionStatus(StrEnum):
     ACTIVE = "active"
     EXPIRED = "expired"
+
+
+class ServiceSignatureStatus(StrEnum):
+    ACTIVE = "active"
+    INACTIVE = "inactive"
+
+
+class LinkCodeStatus(StrEnum):
+    ACTIVE = "active"
+    USED = "used"
+    EXPIRED = "expired"
+
+
+class NotificationType(StrEnum):
+    SEVEN_DAYS = "7d"
+    THREE_DAYS = "3d"
+    ONE_DAY = "1d"
+    EXPIRED = "expired"
+
+
+class NotificationStatus(StrEnum):
+    PENDING = "pending"
+    DISPATCHED = "dispatched"
+    SENT = "sent"
+    FAILED = "failed"

schemas/jwt.py

@@ -3,8 +3,17 @@ from pydantic import BaseModel, Field
 from core.exp import get_exp
 from schemas.providers import ProvidersType
 
+type NumericDate = int | float
 
-class JWTPayload(BaseModel):
+
+class UserJWTPayload(BaseModel):
     sub: str = Field(description="User ID")
     iss: ProvidersType = Field(description="Issuer")
-    exp: float = Field(default_factory=get_exp)
+    exp: NumericDate = Field(default_factory=get_exp)
+
+
+class ServiceJWTPayload(BaseModel):
+    iss: str = Field(description="Service name")
+    iat: NumericDate = Field(description="Issued at")
+    exp: NumericDate = Field(description="Expiration date")
+    acting_as: str = Field(description="Subject of issued call, formatted as `service:id`")

schemas/link_codes.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+
+from pydantic import BaseModel, Field, computed_field
+
+from config import cfg
+
+
+class LinkCodeResponse(BaseModel):
+    code: str = Field()
+    expires_at: datetime = Field()
+
+    @computed_field
+    @property
+    def deep_link(self) -> str:
+        return cfg.bot_url + "?start=" + self.code
+
+
+class LinkCodeConsume(BaseModel):
+    code: str = Field()
+    telegram_id: int = Field()

schemas/login.py

@@ -16,7 +16,7 @@ class UserLoginData(BaseModel):
     telegram: TelegramData | None = None
 
 
-class UserLogin(BaseModel):
+class AuthenticatedUser(BaseModel):
     access_token: str
     refresh_token: str
     expires_at: float

schemas/notifications.py

@@ -0,0 +1,24 @@
+from datetime import UTC, datetime
+
+from pydantic import BaseModel, Field
+
+from schemas.enums import NotificationType
+
+
+class UserNotificationData(BaseModel):
+    notification_id: int = Field()
+    username: str = Field()
+    telegram_id: int | None = Field()
+    expires_at: datetime = Field()
+    notification_type: NotificationType = Field()
+
+
+class NotificationResponse(BaseModel):
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    issued_by: str = Field()
+
+    users: list[UserNotificationData] = Field()
+
+
+class NotificationAcknowledgeRequest(BaseModel):
+    notification_id: int = Field()

schemas/plans.py

@@ -6,6 +6,7 @@ class AddonData(BaseModel):
     name: str
     price: float
     free_threshold: int
+    is_enabled: bool
 
 
 class PricingPlans(BaseModel):

schemas/registration.py

@@ -9,4 +9,6 @@ class UserRegistration(BaseModel):
     username: str | None = Field(None)
     password: str | None = Field(None)
 
+    referal_code: str | None = Field(None)
+
     provider: ProvidersType

schemas/user.py

@@ -1,6 +1,19 @@
+from datetime import datetime
+
 from pydantic import BaseModel, Field
 
 
+class SubscriptionData(BaseModel):
+    has_subscription: bool = Field()
+    devices: int | None = Field(None)
+    expires_at: datetime | None = Field(None)
+    addon_ids: list[str] = Field(default_factory=list)
+    subscription_link: str | None = Field(None)
+    duration_days: int | None = Field(None)
+
+
 class UserInfo(BaseModel):
-    username: str | None = Field()
+    username: str | None = Field(None)
-    telegram_id: str | None = Field()
+    telegram_id: int | None = Field(None)
+    referal_code: str = Field()
+    bonus_balance: float = Field()

services/notifications.py

@@ -0,0 +1,59 @@
+import asyncio
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert
+
+from config import cfg
+from db.models import ServiceNotification, Subscription
+from db.session import UnitOfWork, async_session
+from schemas.enums import NotificationType
+
+
+async def collect_subscription_notifications(uow: UnitOfWork, now: datetime | None = None) -> int:
+    session = uow.session
+    now = now or datetime.now(UTC)
+    periods = (
+        (NotificationType.SEVEN_DAYS, now + timedelta(days=3), now + timedelta(days=7)),
+        (NotificationType.THREE_DAYS, now + timedelta(days=1), now + timedelta(days=3)),
+        (NotificationType.ONE_DAY, now, now + timedelta(days=1)),
+        (NotificationType.EXPIRED, None, now),
+    )
+    created = 0
+
+    for notification_type, starts_at, ends_at in periods:
+        conditions = [Subscription.expires_at <= ends_at]
+        if starts_at is not None:
+            conditions.append(Subscription.expires_at > starts_at)
+
+        subscriptions = await session.scalars(select(Subscription).where(*conditions))
+        rows = [
+            {
+                "subscription_id": subscription.id,
+                "notify_type": notification_type,
+                "sub_expires_at": subscription.expires_at,
+            }
+            for subscription in subscriptions
+        ]
+        if not rows:
+            continue
+
+        stmt = insert(ServiceNotification).values(rows)
+        stmt = stmt.on_conflict_do_nothing(constraint="uq_subscription_notification")
+        result = await session.execute(stmt)
+        created += result.rowcount or 0
+
+    await uow.commit()
+    return created
+
+
+async def run_subscription_notifications() -> None:
+    interval = cfg.notification_scan_interval * 60
+    while True:
+        try:
+            async with async_session() as session, UnitOfWork(session) as uow:
+                await collect_subscription_notifications(uow)
+        except Exception:
+            # A failed iteration must not stop subsequent notification checks.
+            pass
+        await asyncio.sleep(interval)

services/payments.py

@@ -0,0 +1,171 @@
+import hashlib
+import hmac
+import logging
+import math
+from datetime import UTC, datetime
+
+from config import cfg
+from db.models.orders import OrderStatus
+from db.models.transactions import BalanceTxType
+from db.session import UnitOfWork
+from repositories import AddonsRepository
+from repositories.invoices import InvoiceRepository
+from repositories.orders import OrderRepository
+from repositories.pricing import PricingRepository
+from repositories.users import UserRepository
+from schemas.invoices import InvoiceStatus
+from services.plans import get_pricing_model
+from services.rw_sync import enqueue_rw_sync
+from services.subscriptions import (
+    apply_order_now,
+    deduct_order_balance,
+    queue_order_for_later,
+    should_apply_immediately,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def validate_pally_signature(out_sum: str, inv_id: str, signature_value: str) -> bool:
+    raw_string = f"{out_sum}:{inv_id}:{cfg.pally_token}"
+    expected_signature = hashlib.md5(raw_string.encode("utf-8")).hexdigest().upper()
+    return hmac.compare_digest(signature_value, expected_signature)
+
+
+async def process_subscription_purchase(  # noqa: PLR0911, PLR0912
+    uow: UnitOfWork,
+    *,
+    invoice_id: int,
+    trs_id: str,
+    amount: int,
+) -> None:
+    session = uow.session
+    invoice_repo = InvoiceRepository(uow)
+    orders_repo = OrderRepository(uow)
+    users_repo = UserRepository(uow)
+    pricing_repo = PricingRepository(uow)
+
+    invoice = await invoice_repo.get_by_id(invoice_id)
+
+    if not invoice:
+        logger.critical("Bill %s not found in database for TrsId %s", invoice_id, trs_id)
+        return
+
+    if invoice.status != InvoiceStatus.ACTIVE:
+        logger.warning(
+            "Bill %s (TrsId: %s) is already processed with status: %s",
+            invoice.id,
+            trs_id,
+            invoice.status,
+        )
+        return
+
+    if invoice.amount != amount:
+        logger.error(
+            "Amount mismatch for bill %s (TrsId: %s) - Expected: %s, Received: %s",
+            invoice.id,
+            trs_id,
+            invoice.amount,
+            amount,
+        )
+        return
+
+    if invoice.order_id is None:
+        logger.critical("Invoice %s has no linked order for TrsId %s", invoice.id, trs_id)
+        return
+
+    order = await orders_repo.get_by_id(invoice.order_id)
+    if order is None:
+        logger.critical(
+            "Order not found for bill %s (TrsId: %s, user_id=%s, order_id=%s)",
+            invoice.id,
+            trs_id,
+            invoice.creator_id,
+            invoice.order_id,
+        )
+        return
+
+    if order.user_id != invoice.creator_id:
+        logger.critical(
+            "Order %s does not belong to invoice creator %s for TrsId %s",
+            order.id,
+            invoice.creator_id,
+            trs_id,
+        )
+        return
+
+    user = await users_repo.get_user_by_id(invoice.creator_id)
+    if user is None:
+        logger.critical("User %s not found for TrsId %s", invoice.creator_id, trs_id)
+        return
+
+    subscription = user.subscription
+    now = datetime.now(UTC)
+    pricing = await get_pricing_model(AddonsRepository(uow), pricing_repo)
+
+    logger.info(
+        "Processing payment: bill_id=%s, order_id=%s, user_id=%s, amount=%s",
+        invoice.id,
+        order.id,
+        invoice.creator_id,
+        amount,
+    )
+
+    order.status = OrderStatus.PAID
+    invoice.status = InvoiceStatus.PAID
+
+    if order.balance_amount > 0:
+        await deduct_order_balance(
+            session,
+            user=user,
+            order=order,
+            description=f"order {order.id} partial payment from balance",
+        )
+
+    if should_apply_immediately(
+        subscription=subscription,
+        order=order,
+        pricing=pricing,
+        now=now,
+    ):
+        applied_subscription = await apply_order_now(
+            session, user=user, order=order, pricing=pricing, now=now
+        )
+        await enqueue_rw_sync(session, applied_subscription.id)
+    else:
+        if subscription is None:
+            logger.critical(
+                "Cannot queue order %s without subscription for user %s", order.id, user.id
+            )
+            return
+        await queue_order_for_later(order=order, subscription=subscription, now=now)
+
+    referal_id = user.referal_id
+    if referal_id is not None:
+        referal_amount = math.floor(amount * (cfg.referal_bonus / 100))
+        if referal_amount > 0:
+            referal_user = await users_repo.get_user_by_id(referal_id)
+            if referal_user is not None:
+                await users_repo.increase_balance(
+                    referal_id,
+                    referal_amount,
+                    BalanceTxType.REFERRAL_BONUS,
+                    f"referral reward for user {invoice.creator_id} (TrsId: {trs_id})",
+                )
+
+                logger.info(
+                    "Referral bonus processed: referrer_id=%s, amount=%s",
+                    referal_id,
+                    referal_amount,
+                )
+            else:
+                logger.warning(
+                    "Referrer %s not found for user_id=%s while processing TrsId %s",
+                    referal_id,
+                    invoice.creator_id,
+                    trs_id,
+                )
+
+    await uow.commit()
+
+    logger.info("Bill %s marked as PAID for TrsId %s", invoice_id, trs_id)

services/plans.py

@@ -11,7 +11,13 @@ async def get_pricing_model(addons_repo: AddonsRepository, pricing_repo: Pricing
 
     addons = await addons_repo.get_all()
     addons_data = [
-        AddonData(id=a.id, name=a.name, price=a.price, free_threshold=a.free_threshold)
+        AddonData(
+            id=a.id,
+            name=a.name,
+            price=a.price,
+            free_threshold=a.free_threshold,
+            is_enabled=a.is_enabled,
+        )
         for a in addons
     ]
 

services/rw_sync.py

@@ -0,0 +1,126 @@
+import asyncio
+import logging
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+
+from sqlalchemy import delete, select, update
+from sqlalchemy.dialects.postgresql import insert
+
+from config import cfg
+from db.models import RWSyncOutbox, Subscription
+from db.session import UnitOfWork, async_session
+from external.rw import sync_subscription_by_telegram_id
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+LOCK_DURATION = timedelta(minutes=5)
+MAX_RETRY_DELAY = timedelta(hours=1)
+
+
+async def enqueue_rw_sync(session: "AsyncSession", subscription_id: int) -> None:
+    now = datetime.now(UTC)
+    stmt = insert(RWSyncOutbox).values(
+        subscription_id=subscription_id,
+        revision=1,
+        next_attempt_at=now,
+    )
+    await session.execute(
+        stmt.on_conflict_do_update(
+            index_elements=[RWSyncOutbox.subscription_id],
+            set_={
+                "revision": RWSyncOutbox.revision + 1,
+                "next_attempt_at": now,
+                "locked_until": None,
+            },
+        )
+    )
+
+
+async def enqueue_all_rw_syncs(session: "AsyncSession") -> int:
+    subscription_ids = await session.scalars(select(Subscription.id))
+    count = 0
+    for subscription_id in subscription_ids:
+        await enqueue_rw_sync(session, subscription_id)
+        count += 1
+    return count
+
+
+async def process_rw_syncs(uow: UnitOfWork, batch_size: int = 50) -> int:
+    session = uow.session
+    now = datetime.now(UTC)
+    jobs = await session.scalars(
+        select(RWSyncOutbox)
+        .where(
+            RWSyncOutbox.next_attempt_at <= now,
+            (RWSyncOutbox.locked_until.is_(None)) | (RWSyncOutbox.locked_until <= now),
+        )
+        .order_by(RWSyncOutbox.next_attempt_at)
+        .with_for_update(skip_locked=True)
+        .limit(batch_size)
+    )
+    jobs = list(jobs)
+    for job in jobs:
+        job.locked_until = now + LOCK_DURATION
+    await uow.commit()
+
+    for job in jobs:
+        subscription = job.subscription
+        user = subscription.user
+        synced = await sync_subscription_by_telegram_id(
+            expires_at=subscription.expires_at,
+            devices=subscription.devices,
+            telegram_id=user.telegram_id,
+            username=user.username,
+        )
+        if synced:
+            await session.execute(
+                delete(RWSyncOutbox).where(
+                    RWSyncOutbox.id == job.id,
+                    RWSyncOutbox.revision == job.revision,
+                )
+            )
+            await uow.commit()
+            continue
+
+        delay = min(timedelta(minutes=2**job.attempts), MAX_RETRY_DELAY)
+        await session.execute(
+            update(RWSyncOutbox)
+            .where(RWSyncOutbox.id == job.id, RWSyncOutbox.revision == job.revision)
+            .values(
+                attempts=RWSyncOutbox.attempts + 1,
+                next_attempt_at=datetime.now(UTC) + delay,
+                locked_until=None,
+            )
+        )
+        await uow.commit()
+        logger.warning("RW sync failed for subscription_id=%s", subscription.id)
+
+    return len(jobs)
+
+
+async def run_rw_sync_worker() -> None:
+    interval = cfg.rw_sync_interval_minutes * 60
+    while True:
+        try:
+            async with async_session() as session, UnitOfWork(session) as uow:
+                await process_rw_syncs(uow)
+        except Exception:
+            logger.exception("RW sync worker iteration failed")
+        await asyncio.sleep(interval)
+
+
+async def run_rw_sync_reconciler() -> None:
+    interval = cfg.rw_sync_reconcile_interval_minutes * 60
+    while True:
+        try:
+            async with async_session() as session:
+                async with UnitOfWork(session) as uow:
+                    count = await enqueue_all_rw_syncs(uow.session)
+                    await uow.commit()
+                logger.info("Queued %s subscriptions for RW reconciliation", count)
+        except Exception:
+            logger.exception("RW reconciliation iteration failed")
+        await asyncio.sleep(interval)

services/subscriptions.py

@@ -6,9 +6,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from db.models import Subscription, SubscriptionAddon, User
 from db.models.orders import Order, OrderStatus
 from db.models.transactions import BalanceTransaction, BalanceTxType
-from external.rw import sync_subscription_by_telegram_id
 from schemas.enums import SubscriptionStatus
 from schemas.plans import PricingPlans
+from services.rw_sync import enqueue_rw_sync
 
 
 def calculate_plan_monthly_price(
@@ -72,7 +72,7 @@ async def replace_subscription_addons(
 
 
 async def ensure_subscription(
-    user: User, session: AsyncSession, starts_at: datetime
+    user: User, session: AsyncSession, starts_at: datetime, duration_days: int
 ) -> Subscription:
     subscription = user.subscription
     if subscription is not None:
@@ -83,6 +83,7 @@ async def ensure_subscription(
         devices=0,
         status=SubscriptionStatus.EXPIRED,
         expires_at=starts_at,
+        duration_days=duration_days,
     )
     session.add(subscription)
     await session.flush()
@@ -121,7 +122,7 @@ async def apply_order_now(
     pricing: PricingPlans,
     now: datetime,
 ) -> Subscription:
-    subscription = await ensure_subscription(user, session, now)
+    subscription = await ensure_subscription(user, session, now, order.duration_days)
     addon_ids = [addon.addon_id for addon in order.addons]
     current_addons = [] if subscription.devices == 0 else get_subscription_addon_ids(subscription)
 
@@ -193,9 +194,4 @@ async def sync_user_subscription(
             subscription.status = SubscriptionStatus.EXPIRED
 
     if applied_due_orders:
-        await sync_subscription_by_telegram_id(
+        await enqueue_rw_sync(session, subscription.id)
-            expires_at=subscription.expires_at,
-            devices=subscription.devices,
-            telegram_id=user.telegram_id,
-            username=user.username,
-        )

services/users.py

@@ -1,23 +1,28 @@
 from core.secrets import generate_pair, hash_refresh_token
 from db.models.users import User
 from repositories.sessions import SessionsRepository
-from schemas.login import UserLogin
+from schemas.login import AuthenticatedUser
 from schemas.providers import ProvidersType
 from schemas.user import UserInfo
 
 
 async def authorize_user(
     sessions_repo: SessionsRepository, user: User, iss: ProvidersType
-) -> UserLogin:
+) -> AuthenticatedUser:
     key_pair = generate_pair(user.id, iss)
 
     refresh_token_hash = hash_refresh_token(key_pair.refresh_token)
 
     await sessions_repo.create(user_id=user.id, refresh_token_hash=refresh_token_hash, iss=iss)
 
-    return UserLogin(
+    return AuthenticatedUser(
         access_token=key_pair.access_token,
         refresh_token=key_pair.refresh_token,
-        user=UserInfo(username=user.username, telegram_id=user.telegram_id),
+        user=UserInfo(
+            username=user.username,
+            telegram_id=user.telegram_id,
+            referal_code=user.referal_code,
+            bonus_balance=user.balance,
+        ),
         expires_at=key_pair.expires_at,
     )

tests/conftest.py

@@ -0,0 +1,17 @@
+import sys
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from main import app
+
+
+@pytest.fixture
+def client():
+    app.dependency_overrides.clear()
+    with TestClient(app, raise_server_exceptions=False) as test_client:
+        yield test_client
+    app.dependency_overrides.clear()

tests/test_auth_routes.py

@@ -0,0 +1,127 @@
+# ruff: noqa: PLR2004
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+
+def test_signup_rejects_missing_credentials(client):
+    response = client.post("/auth/signup", json={"provider": "credentials"})
+
+    assert response.status_code == 400
+    assert response.json()["detail"] == "Username or password is not provided"
+
+
+def test_signup_rejects_unsupported_provider(client):
+    response = client.post(
+        "/auth/signup", json={"provider": "telegram", "username": "alice", "password": "Strong123!"}
+    )
+
+    assert response.status_code == 400
+    assert response.json()["detail"] == "Unsupported provider"
+
+
+def test_signup_rejects_existing_username(client):
+    repository = SimpleNamespace(get_user_by_username=AsyncMock(return_value=object()))
+    with patch("routes.auth.UserRepository", return_value=repository):
+        response = client.post(
+            "/auth/signup",
+            json={"provider": "credentials", "username": "alice", "password": "Strong123!"},
+        )
+
+    assert response.status_code == 409
+    assert response.json()["detail"] == "User already exists"
+
+
+def test_signup_rejects_weak_password(client):
+    repository = SimpleNamespace(get_user_by_username=AsyncMock(return_value=None))
+    with (
+        patch("routes.auth.UserRepository", return_value=repository),
+        patch("routes.auth.estimate_password_strength", return_value=False),
+    ):
+        response = client.post(
+            "/auth/signup",
+            json={"provider": "credentials", "username": "alice", "password": "weak"},
+        )
+
+    assert response.status_code == 422
+    assert response.json()["detail"] == "Password is not secure."
+
+
+def test_signup_accepts_unknown_referral_code(client):
+    created_user = SimpleNamespace(
+        id=1, username="alice", telegram_id=None, referal_code="new-code", balance=0
+    )
+    repository = SimpleNamespace(
+        get_user_by_username=AsyncMock(return_value=None),
+        get_user_by_ref_code=AsyncMock(return_value=None),
+        create=AsyncMock(return_value=created_user),
+    )
+    sessions_repository = SimpleNamespace(create=AsyncMock(return_value=None))
+    with (
+        patch("routes.auth.UserRepository", return_value=repository),
+        patch("routes.auth.SessionsRepository", return_value=sessions_repository),
+        patch("routes.auth.estimate_password_strength", return_value=True),
+        patch("routes.auth.hash_password", return_value="hashed"),
+    ):
+        response = client.post(
+            "/auth/signup",
+            json={
+                "provider": "credentials",
+                "username": "alice",
+                "password": "Strong123!",
+                "referal_code": "unknown",
+            },
+        )
+
+    assert response.status_code == 200
+    repository.create.assert_awaited_once_with(
+        username="alice", hashed_password="hashed", referal_id=None
+    )
+    assert response.json()["user"]["referal_code"] == "new-code"
+
+
+def test_login_distinguishes_unknown_user_and_bad_password(client):
+    unknown_repository = SimpleNamespace(get_user_by_username=AsyncMock(return_value=None))
+    with patch("routes.auth.UserRepository", return_value=unknown_repository):
+        unknown_response = client.post(
+            "/auth/login",
+            json={"provider": "credentials", "username": "alice", "password": "secret"},
+        )
+
+    existing_user = SimpleNamespace(hashed_password="hash")
+    existing_repository = SimpleNamespace(
+        get_user_by_username=AsyncMock(return_value=existing_user)
+    )
+    with (
+        patch("routes.auth.UserRepository", return_value=existing_repository),
+        patch("routes.auth.SessionsRepository"),
+        patch("routes.auth.verify_password", return_value=False),
+    ):
+        password_response = client.post(
+            "/auth/login",
+            json={"provider": "credentials", "username": "alice", "password": "secret"},
+        )
+
+    assert unknown_response.status_code == password_response.status_code == 401
+    assert unknown_response.json()["detail"] == "User doesn't exist."
+    assert password_response.json()["detail"] == "Invalid password"
+
+
+def test_login_telegram_is_explicitly_unavailable(client):
+    response = client.post("/auth/login", json={"provider": "telegram"})
+
+    assert response.status_code == 503
+
+
+def test_refresh_requires_query_parameters_and_rejects_unknown_token(client):
+    missing_response = client.post("/auth/refresh")
+    repository = SimpleNamespace(get_session_by_hash=AsyncMock(return_value=None))
+    with (
+        patch("routes.auth.SessionsRepository", return_value=repository),
+        patch("routes.auth.hash_refresh_token", return_value="token-hash"),
+    ):
+        invalid_response = client.post("/auth/refresh?refresh_token=expired&iss=credentials")
+
+    assert missing_response.status_code == 422
+    assert invalid_response.status_code == 401
+    assert invalid_response.json()["detail"] == "Refresh token is invalid."

tests/test_link_code_routes.py

@@ -0,0 +1,137 @@
+# ruff: noqa: PLR2004
+
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from config import cfg
+from routes import link_codes
+from schemas.enums import LinkCodeStatus
+
+
+def auth_context(user_id=42):
+    return SimpleNamespace(user=SimpleNamespace(id=user_id))
+
+
+def service_identity():
+    return SimpleNamespace(service="telegram-bot")
+
+
+def test_generate_link_code_requires_authorization(client):
+    response = client.post("/link-codes")
+
+    assert response.status_code == 403
+    assert response.json()["detail"] == "No authorization provided."
+
+
+def test_generate_link_code_creates_active_code_with_expiry_and_deep_link(client):
+    client.app.dependency_overrides[link_codes.get_auth_context] = auth_context
+    created_codes = []
+
+    async def create_code(session, *, code, user_id, status, expires_at):
+        created_codes.append(
+            SimpleNamespace(
+                session=session,
+                code=code,
+                user_id=user_id,
+                status=status,
+                expires_at=expires_at,
+            )
+        )
+        return created_codes[-1]
+
+    before = datetime.now(UTC)
+    with (
+        patch(
+            "routes.link_codes.secrets.token_urlsafe", return_value="one-time-code"
+        ) as token_urlsafe,
+        patch("routes.link_codes.create_link_code", side_effect=create_code),
+    ):
+        response = client.post("/link-codes")
+    after = datetime.now(UTC)
+
+    assert response.status_code == 201
+    assert response.json() == {
+        "code": "one-time-code",
+        "expires_at": created_codes[0].expires_at.isoformat().replace("+00:00", "Z"),
+        "deep_link": f"{cfg.bot_url}?start=one-time-code",
+    }
+    assert created_codes[0].user_id == 42
+    assert created_codes[0].status is LinkCodeStatus.ACTIVE
+    token_urlsafe.assert_called_once_with(cfg.link_code_length)
+    assert before + timedelta(minutes=cfg.link_code_ttl) <= created_codes[0].expires_at
+    assert created_codes[0].expires_at <= after + timedelta(minutes=cfg.link_code_ttl)
+
+
+def test_consume_link_code_requires_code_and_telegram_id(client):
+    client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
+
+    missing_code = client.post("/link-codes/consume", json={"telegram_id": 12345})
+    missing_telegram_id = client.post("/link-codes/consume", json={"code": "link-code"})
+    invalid_telegram_id = client.post(
+        "/link-codes/consume", json={"code": "link-code", "telegram_id": "not-an-id"}
+    )
+
+    assert missing_code.status_code == 422
+    assert missing_telegram_id.status_code == 422
+    assert invalid_telegram_id.status_code == 422
+
+
+def test_consume_link_code_rejects_unknown_code(client):
+    client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
+    lookup = AsyncMock(return_value=None)
+    with patch("routes.link_codes.get_link_code_by_code", lookup):
+        response = client.post(
+            "/link-codes/consume", json={"code": "missing", "telegram_id": 12345}
+        )
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Code not found"
+    lookup.assert_awaited_once()
+    assert lookup.await_args.args[1] == "missing"
+
+
+def test_consume_link_code_rejects_code_for_deleted_user(client):
+    client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
+    link_code = SimpleNamespace(user_id=42, status=LinkCodeStatus.ACTIVE)
+    repository = SimpleNamespace(get_user_by_id=AsyncMock(return_value=None))
+    with (
+        patch("routes.link_codes.get_link_code_by_code", new=AsyncMock(return_value=link_code)),
+        patch("routes.link_codes.UserRepository", return_value=repository),
+    ):
+        response = client.post(
+            "/link-codes/consume", json={"code": "orphaned", "telegram_id": 12345}
+        )
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "User not found"
+    repository.get_user_by_id.assert_awaited_once_with(42)
+
+
+def test_consume_link_code_updates_telegram_id_and_returns_user(client):
+    client.app.dependency_overrides[link_codes.get_service_identity] = service_identity
+    link_code = SimpleNamespace(user_id=42, status=LinkCodeStatus.ACTIVE)
+    user = SimpleNamespace(
+        username="alice", telegram_id=12345, referal_code="ref-code", balance=100
+    )
+    repository = SimpleNamespace(
+        get_user_by_id=AsyncMock(return_value=user),
+        update_telegram_id=AsyncMock(return_value=user),
+    )
+    with (
+        patch("routes.link_codes.get_link_code_by_code", new=AsyncMock(return_value=link_code)),
+        patch("routes.link_codes.UserRepository", return_value=repository),
+    ):
+        response = client.post(
+            "/link-codes/consume", json={"code": "link-code", "telegram_id": 12345}
+        )
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "username": "alice",
+        "telegram_id": 12345,
+        "referal_code": "ref-code",
+        "bonus_balance": 100,
+    }
+    repository.get_user_by_id.assert_awaited_once_with(42)
+    repository.update_telegram_id.assert_awaited_once_with(user, 12345)

tests/test_notifications.py

@@ -0,0 +1,47 @@
+import asyncio
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+from schemas.enums import NotificationType
+from services.notifications import collect_subscription_notifications
+
+
+class FakeUnitOfWork:
+    def __init__(self, subscriptions_by_period):
+        self.session = self
+        self.subscriptions_by_period = iter(subscriptions_by_period)
+        self.executed = []
+        self.committed = False
+
+    async def scalars(self, _statement):
+        return next(self.subscriptions_by_period)
+
+    async def execute(self, statement):
+        self.executed.append(statement)
+        return SimpleNamespace(rowcount=1)
+
+    async def commit(self):
+        self.committed = True
+
+
+def test_collects_notification_for_each_expiry_period():
+    now = datetime(2026, 8, 20, tzinfo=UTC)
+    uow = FakeUnitOfWork(
+        [
+            [SimpleNamespace(id=1, expires_at=now + timedelta(days=6))],
+            [SimpleNamespace(id=2, expires_at=now + timedelta(days=2))],
+            [SimpleNamespace(id=3, expires_at=now + timedelta(hours=12))],
+            [SimpleNamespace(id=4, expires_at=now - timedelta(hours=1))],
+        ]
+    )
+
+    created = asyncio.run(collect_subscription_notifications(uow, now))
+
+    assert created == len(uow.executed)
+    assert uow.committed
+    assert [statement.compile().params["notify_type_m0"] for statement in uow.executed] == [
+        NotificationType.SEVEN_DAYS,
+        NotificationType.THREE_DAYS,
+        NotificationType.ONE_DAY,
+        NotificationType.EXPIRED,
+    ]

tests/test_services_and_payments.py

@@ -0,0 +1,106 @@
+# ruff: noqa: PLR2004
+
+import hashlib
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from external.pally import BillStatus
+from schemas.plans import AddonData, PricingPlans
+from services.payments import validate_pally_signature
+from services.subscriptions import calculate_order_total, should_apply_immediately
+
+
+def pricing():
+    return PricingPlans(
+        device_price=100,
+        addons=[
+            AddonData(id="a", name="A", price=25, free_threshold=0, is_enabled=True),
+            AddonData(id="b", name="B", price=50, free_threshold=0, is_enabled=True),
+        ],
+    )
+
+
+def test_price_calculation_handles_fractional_months_and_unknown_addons():
+    assert calculate_order_total(pricing(), devices=2, addon_ids=["a"], duration_days=15) == 112.5
+
+    with pytest.raises(KeyError):
+        calculate_order_total(pricing(), devices=1, addon_ids=["missing"], duration_days=30)
+
+
+def test_should_apply_immediately_distinguishes_upgrade_from_downgrade():
+    now = datetime.now(UTC)
+    subscription = SimpleNamespace(
+        devices=2,
+        expires_at=now + timedelta(days=10),
+        addons=[SimpleNamespace(addon_id="a")],
+    )
+    upgrade = SimpleNamespace(
+        devices=3, addons=[SimpleNamespace(addon_id="a"), SimpleNamespace(addon_id="b")]
+    )
+    downgrade = SimpleNamespace(devices=1, addons=[SimpleNamespace(addon_id="a")])
+
+    assert should_apply_immediately(
+        order=upgrade, subscription=subscription, pricing=pricing(), now=now
+    )
+    assert not should_apply_immediately(
+        order=downgrade, subscription=subscription, pricing=pricing(), now=now
+    )
+    assert should_apply_immediately(order=downgrade, subscription=None, pricing=pricing(), now=now)
+
+
+def test_pally_signature_is_exact_and_case_sensitive():
+    signature = hashlib.md5(b"10:42:test-token").hexdigest().upper()
+    with patch("services.payments.cfg.pally_token", "test-token"):
+        assert validate_pally_signature("10", "42", signature)
+        assert not validate_pally_signature("10", "42", signature.lower())
+        assert not validate_pally_signature("11", "42", signature)
+
+
+def callback_data(**overrides):
+    data = {
+        "InvId": "42",
+        "OutSum": "10",
+        "Commission": "0",
+        "TrsId": "transaction",
+        "Status": BillStatus.SUCCESS,
+        "CurrencyIn": "RUB",
+        "SignatureValue": "valid",
+    }
+    data.update(overrides)
+    return data
+
+
+def test_callback_rejects_invalid_signature_before_processing(client):
+    with patch("routes.payments.pally.validate_pally_signature", return_value=False):
+        response = client.post("/payments/pally/result", data=callback_data())
+
+    assert response.status_code == 403
+    assert response.json()["detail"] == "Invalid signature."
+
+
+def test_callback_acknowledges_valid_non_success_and_non_numeric_invoice(client):
+    with patch("routes.payments.pally.validate_pally_signature", return_value=True):
+        failed_response = client.post("/payments/pally/result", data=callback_data(Status="Failed"))
+        invalid_id_response = client.post(
+            "/payments/pally/result", data=callback_data(InvId="order-42")
+        )
+
+    assert failed_response.status_code == invalid_id_response.status_code == 200
+    assert failed_response.json() == invalid_id_response.json() == "OK"
+
+
+def test_callback_uses_balance_amount_when_present(client):
+    process = AsyncMock()
+    with (
+        patch("routes.payments.pally.validate_pally_signature", return_value=True),
+        patch("routes.payments.pally.process_subscription_purchase", process),
+    ):
+        response = client.post("/payments/pally/result", data=callback_data(BalanceAmount="7.9"))
+
+    assert response.status_code == 200
+    process.assert_awaited_once()
+    assert process.await_args.kwargs["invoice_id"] == 42
+    assert process.await_args.kwargs["amount"] == 7

tests/test_user_and_plan_routes.py

@@ -0,0 +1,131 @@
+# ruff: noqa: PLR2004, PLW0108
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from routes import users
+
+
+def auth_context(subscription=None):
+    return SimpleNamespace(
+        user=SimpleNamespace(
+            id=1,
+            username="alice",
+            telegram_id=12345,
+            referal_code="ref-code",
+            balance=12.5,
+            subscription=subscription,
+        )
+    )
+
+
+def test_protected_user_endpoint_requires_authorization(client):
+    response = client.get("/users/me")
+
+    assert response.status_code == 403
+
+
+def test_get_me_serializes_auth_context(client):
+    client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
+
+    response = client.get("/users/me")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "username": "alice",
+        "telegram_id": 12345,
+        "referal_code": "ref-code",
+        "bonus_balance": 12.5,
+    }
+
+
+def test_subscription_without_local_subscription_has_empty_details(client):
+    client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
+
+    response = client.get("/users/subscription")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "has_subscription": False,
+        "devices": None,
+        "expires_at": None,
+        "addon_ids": [],
+        "subscription_link": None,
+        "duration_days": None,
+    }
+
+
+def test_subscription_keeps_local_data_when_external_user_is_missing(client):
+    subscription = SimpleNamespace(
+        id=2,
+        devices=3,
+        expires_at=datetime(2030, 1, 1, tzinfo=UTC),
+        duration_days=30,
+        addons=[SimpleNamespace(addon_id="a"), SimpleNamespace(addon_id="b")],
+    )
+    client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context(subscription)
+    with (
+        patch("routes.users.get_sdk", return_value="sdk"),
+        patch("routes.users.get_rw_user", new=AsyncMock(return_value=None)),
+    ):
+        response = client.get("/users/subscription")
+
+    assert response.status_code == 200
+    assert response.json()["has_subscription"] is True
+    assert response.json()["addon_ids"] == ["a", "b"]
+    assert response.json()["subscription_link"] is None
+
+
+def test_hwid_returns_empty_list_and_rejects_missing_external_user(client):
+    client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
+    with (
+        patch("routes.users.get_sdk", return_value="sdk"),
+        patch("routes.users.get_rw_user", new=AsyncMock(return_value=None)),
+    ):
+        missing_user = client.get("/users/subscription/hwid")
+
+    rw_user = SimpleNamespace(uuid="uuid")
+    with (
+        patch("routes.users.get_sdk", return_value="sdk"),
+        patch("routes.users.get_rw_user", new=AsyncMock(return_value=rw_user)),
+        patch("routes.users.get_hwid_list", new=AsyncMock(return_value=None)),
+    ):
+        empty_list = client.get("/users/subscription/hwid")
+
+    assert missing_user.status_code == 403
+    assert empty_list.status_code == 200
+    assert empty_list.json() == []
+
+
+def test_hwid_maps_client_prefix_and_delete_preserves_external_result(client):
+    client.app.dependency_overrides[users.get_auth_context] = lambda: auth_context()
+    rw_user = SimpleNamespace(uuid="uuid")
+    device = SimpleNamespace(
+        platform="Windows", device_model="PC", user_agent="v2ray/6.0", hwid="abc"
+    )
+    with (
+        patch("routes.users.get_sdk", return_value="sdk"),
+        patch("routes.users.get_rw_user", new=AsyncMock(return_value=rw_user)),
+        patch("routes.users.get_hwid_list", new=AsyncMock(return_value=[device])),
+        patch("routes.users.delete_hwid", new=AsyncMock(return_value=False)),
+    ):
+        devices_response = client.get("/users/subscription/hwid")
+        delete_response = client.delete("/users/subscription/hwid?hwid=abc")
+        missing_hwid_response = client.delete("/users/subscription/hwid")
+
+    assert devices_response.json() == [
+        {"os": "Windows", "model": "PC", "client": "v2ray", "hwid": "abc"}
+    ]
+    assert delete_response.status_code == 200
+    assert delete_response.json() == {"success": False}
+    assert missing_hwid_response.status_code == 422
+
+
+def test_plans_returns_pricing_service_result(client):
+    expected = {"device_price": 100, "addons": []}
+    with patch("routes.plans.get_pricing_model", new=AsyncMock(return_value=expected)):
+        response = client.get("/plans/")
+
+    assert response.status_code == 200
+    assert response.json() == expected