fix: replaced remnawave_sdk to local gitea

.gitignore

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

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,28 +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()
 
-    private_key_fp: str = Field()
-    public_key_fp: str = Field()
-
-    access_token_ttl: int = Field(description="Access token TTL (minutes)")
-
-    pally_shop_id: str = Field()
-    pally_token: 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)
+
+    min_password_length: int = Field(8)
+    link_code_length: int = Field(8)
+    password_security_threshold: int = Field(2)
 
     @computed_field
     @property
@@ -54,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:
-    content = decode_jwt(token)
+async def authorize_bot(kid: str, token: str, uow: UnitOfWork) -> AuthContext:
+    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)
-        await sync_user_subscription(session, user=ctx.user)
-        await session.commit()
+        ctx = await jwt.authorize(token, uow)
+        await sync_user_subscription(uow.session, user=ctx.user)
+        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")
-    except jwt.ExpiredSignatureError:
+        header = jwt.get_unverified_header(token)
+        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

@@ -37,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(
-        "User",
-        remote_side=[id],
-    )
+    referal: Mapped["User | None"] = relationship("User", remote_side=[id], lazy="selectin")

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:
-    if telegram_id is not None:
-        return f"tg_{telegram_id}"
-
+def _build_rw_username(*, username: str | None) -> str | None:
     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
@@ -439,7 +436,7 @@ async def get_rw_user(
         rw_user = await get_user_by_telegram_id(sdk, telegram_id)
 
     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:
             logger.warning(
                 "Cannot build username for telegram_id=%s username=%s",
@@ -450,6 +447,7 @@ async def get_rw_user(
 
         rw_user = await get_user_by_username(sdk, rw_username)
         return rw_user
+    return rw_user
 
 
 async def sync_subscription_by_telegram_id(
@@ -465,8 +463,13 @@ async def sync_subscription_by_telegram_id(
         return False
 
     rw_user = await get_rw_user(sdk, telegram_id=telegram_id, username=username)
+    if expires_at <= datetime.now(UTC):
+        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)
         rw_user = await create_user(
             sdk=sdk,
             username=rw_username,
@@ -485,7 +488,8 @@ async def sync_subscription_by_telegram_id(
 
     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):

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:
-        self.session = session
+    def __init__(self, uow: UnitOfWork) -> None:
+        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:
-        self.session = session
+    def __init__(self, uow: UnitOfWork) -> None:
+        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:
-        self.session = session
+    def __init__(self, uow: UnitOfWork) -> None:
+        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:
-        self.session = session
+    def __init__(self, uow: UnitOfWork) -> None:
+        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:
-        self.session = session
+    def __init__(self, uow: UnitOfWork) -> None:
+        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:
-        self.session = session
+    def __init__(self, uow: UnitOfWork) -> None:
+        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)
@@ -44,8 +45,6 @@ class UserRepository:
             referal_id=referal_id,
         )
         self.session.add(obj)
-        await self.session.commit()
-
         return obj
 
     async def increase_balance(
@@ -67,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,11 @@
 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
 
@@ -11,5 +14,8 @@ routers: list[APIRouter] = [
     plans_router,
     orders_router,
     users_routers,
-    *payment_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 db.session import get_db
+from core.secrets import (
+    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")
-async def signup(req: UserRegistration, session: AsyncSession = Depends(get_db)):
-    users_repo = UserRepository(session)
+@router.post("/signup", response_model=AuthenticatedUser)
+async def signup(req: UserRegistration, uow: UnitOfWork = Depends(get_uow)) -> AuthenticatedUser:
+    users_repo = UserRepository(uow)
+    sessions_repo = SessionsRepository(uow)
 
     if req.provider == "credentials":
         if not req.username or not req.password:
@@ -27,6 +31,9 @@ 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)
         referal_id = None
         if req.referal_code:
@@ -36,21 +43,28 @@ async def signup(req: UserRegistration, session: AsyncSession = Depends(get_db))
         user = await users_repo.create(
             username=req.username, hashed_password=password_hash, referal_id=referal_id
         )
-        return JSONResponse(
-            UserInfo(
-                username=user.username,
-                telegram_id=user.telegram_id,
-                referal_code=user.referal_code,
-            ).model_dump(),
-            status_code=201,
+        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)
-async def login(req: UserLoginData, session: AsyncSession = Depends(get_db)):
-    users_repo = UserRepository(session)
-    sessions_repo = SessionsRepository(session)
+@router.post("/login", response_model=AuthenticatedUser)
+async def login(req: UserLoginData, uow: UnitOfWork = Depends(get_uow)):
+    users_repo = UserRepository(uow)
+    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.")
@@ -63,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":
@@ -73,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)):
-    sessions_repo = SessionsRepository(session)
+async def refresh(refresh_token: str, iss: ProvidersType, uow: UnitOfWork = Depends(get_uow)):
+    sessions_repo = SessionsRepository(uow)
 
     token_hash = hash_refresh_token(refresh_token)
     token_entry = await sessions_repo.get_session_by_hash(token_hash)
@@ -83,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)
-    pricing_repo = PricingRepository(session)
-    invoices_repo = InvoiceRepository(session)
-    orders_repo = OrderRepository(session)
+    addons_repo = AddonsRepository(uow)
+    pricing_repo = PricingRepository(uow)
+    invoices_repo = InvoiceRepository(uow)
+    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

@@ -3,13 +3,16 @@ import logging
 
 from fastapi import Depends, Form, HTTPException
 from fastapi.routing import APIRouter
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from core.deps import get_db
+from db.models.transactions import BalanceTxType
+from db.session import UnitOfWork, get_uow
 from external.pally import BillStatus
+from repositories.invoices import InvoiceRepository
+from repositories.users import UserRepository
+from schemas.invoices import InvoiceStatus
 from services.payments import process_subscription_purchase, validate_pally_signature
 
-router = APIRouter(prefix="/payments/pally")
+router = APIRouter(prefix="/pally")
 
 logger = logging.getLogger(__name__)
 
@@ -35,7 +38,7 @@ async def pally_callback(
     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),
 ):
     logger.info(
         "Pally webhook received - InvId: %s, OutSum: %s, Commission: %s, TrsId: %s, Status: %s, "
@@ -82,11 +85,62 @@ async def pally_callback(
 
     amount = int(float(BalanceAmount)) if BalanceAmount is not None else int(float(OutSum))
 
-    await process_subscription_purchase(
-        session,
-        invoice_id=int(invoice_id_str),
-        trs_id=TrsId,
-        amount=amount,
-    )
+    try:
+        await process_subscription_purchase(
+            uow,
+            invoice_id=int(invoice_id_str),
+            trs_id=TrsId,
+            amount=amount,
+        )
+    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()
+
+        invoice_repo = InvoiceRepository(uow)
+        invoice = await invoice_repo.get_by_id(int(invoice_id_str))
+        if invoice is None:
+            logger.critical(
+                "Cannot credit fallback balance: bill %s was not found (TrsId: %s)",
+                invoice_id_str,
+                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(
+                "Cannot credit fallback balance: user %s was not found " "for bill %s (TrsId: %s)",
+                invoice.creator_id,
+                invoice_id_str,
+                TrsId,
+            )
+            raise
+
+        await users_repo.increase_balance(
+            user.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)
+        await uow.commit()
+        logger.info(
+            "Fallback payment credit processed: user_id=%s, amount=%s, " "invoice_id=%s, TrsId=%s",
+            user.id,
+            amount,
+            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)):
-    addons_repo = AddonsRepository(session)
-    pricing_repo = PricingRepository(session)
+async def get_plans(uow: UnitOfWork = Depends(get_uow)):
+    addons_repo = AddonsRepository(uow)
+    pricing_repo = PricingRepository(uow)
 
     res = await get_pricing_model(addons_repo, pricing_repo)
     return res

routes/users.py

@@ -1,3 +1,5 @@
+import logging
+
 from fastapi import APIRouter, Depends, HTTPException
 
 from core.deps import get_auth_context
@@ -8,6 +10,7 @@ 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)
@@ -16,6 +19,7 @@ async def get_me(ctx: AuthContext = Depends(get_auth_context)):
         username=ctx.user.username,
         telegram_id=ctx.user.telegram_id,
         referal_code=ctx.user.referal_code,
+        bonus_balance=ctx.user.balance,
     )
 
 
@@ -24,18 +28,37 @@ 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.id for a in sub.addons],
+        addon_ids=[a.addon_id for a in sub.addons],
         subscription_link=build_subscription_link(rw_user.short_uuid),
+        duration_days=sub.duration_days,
     )
 
 

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/user.py

@@ -1,21 +1,19 @@
 from datetime import datetime
 
-from pydantic import BaseModel, Field, computed_field
+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)
-
-    @property
-    @computed_field
-    def has_subscription(self) -> bool:
-        return all([self.devices, self.expires_at, self.subscription_link, self.addon_ids])
+    duration_days: int | None = Field(None)
 
 
 class UserInfo(BaseModel):
     username: str | None = Field(None)
-    telegram_id: str | None = Field(None)
+    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

@@ -6,8 +6,8 @@ from datetime import UTC, datetime
 
 from config import cfg
 from db.models.orders import OrderStatus
-from db.models.transactions import BalanceTransaction, BalanceTxType
-from external.rw import sync_subscription_by_telegram_id
+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
@@ -15,6 +15,7 @@ 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,
@@ -32,16 +33,17 @@ def validate_pally_signature(out_sum: str, inv_id: str, signature_value: str) ->
 
 
 async def process_subscription_purchase(  # noqa: PLR0911, PLR0912
-    session,
+    uow: UnitOfWork,
     *,
     invoice_id: int,
     trs_id: str,
     amount: int,
 ) -> None:
-    invoice_repo = InvoiceRepository(session)
-    orders_repo = OrderRepository(session)
-    users_repo = UserRepository(session)
-    pricing_repo = PricingRepository(session)
+    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)
 
@@ -99,7 +101,7 @@ async def process_subscription_purchase(  # noqa: PLR0911, PLR0912
 
     subscription = user.subscription
     now = datetime.now(UTC)
-    pricing = await get_pricing_model(AddonsRepository(session), pricing_repo)
+    pricing = await get_pricing_model(AddonsRepository(uow), pricing_repo)
 
     logger.info(
         "Processing payment: bill_id=%s, order_id=%s, user_id=%s, amount=%s",
@@ -129,12 +131,7 @@ async def process_subscription_purchase(  # noqa: PLR0911, PLR0912
         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,
-        )
+        await enqueue_rw_sync(session, applied_subscription.id)
     else:
         if subscription is None:
             logger.critical(
@@ -147,19 +144,14 @@ async def process_subscription_purchase(  # noqa: PLR0911, PLR0912
     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)
+            referal_user = await users_repo.get_user_by_id(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: {trs_id})",
-                    )
+                await users_repo.increase_balance(
+                    referal_id,
+                    referal_amount,
+                    BalanceTxType.REFERRAL_BONUS,
+                    f"referral reward for user {invoice.creator_id} (TrsId: {trs_id})",
                 )
-                referal_user.balance += referal_amount
 
                 logger.info(
                     "Referral bonus processed: referrer_id=%s, amount=%s",
@@ -174,6 +166,6 @@ async def process_subscription_purchase(  # noqa: PLR0911, PLR0912
                     trs_id,
                 )
 
-    await session.commit()
+    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(
-            expires_at=subscription.expires_at,
-            devices=subscription.devices,
-            telegram_id=user.telegram_id,
-            username=user.username,
-        )
+        await enqueue_rw_sync(session, subscription.id)

services/users.py

@@ -1,27 +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,
             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