feat(sub): handling upgrade/downgrade

rw integration and /me endpoints are left

alembic/versions/0ea5625b4913_order_durationdays.py

@@ -0,0 +1,46 @@
+"""+order.durationdays
+
+Revision ID: 0ea5625b4913
+Revises: b97ce5b7d663
+Create Date: 2026-08-02 13:17:18.429961
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = '0ea5625b4913'
+down_revision: Union[str, Sequence[str], None] = 'b97ce5b7d663'
+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! ###
+    orderstatus = postgresql.ENUM('PENDING', 'PAID', 'DISCARDED', name='orderstatus')
+    orderstatus.create(op.get_bind())
+
+    op.create_unique_constraint(None, 'order_addons', ['id'])
+    op.add_column('orders', sa.Column('duration_days', sa.INTEGER(), nullable=False))
+    op.add_column('orders', sa.Column('status', sa.Enum('PENDING', 'PAID', 'DISCARDED', name='orderstatus'), nullable=False))
+    op.create_unique_constraint(None, 'orders', ['id'])
+    op.create_unique_constraint(None, 'subscription_addons', ['id'])
+    op.create_unique_constraint(None, 'subscriptions', ['id'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'subscriptions', type_='unique')
+    op.drop_constraint(None, 'subscription_addons', type_='unique')
+    op.drop_constraint(None, 'orders', type_='unique')
+    op.drop_column('orders', 'status')
+    op.drop_column('orders', 'duration_days')
+    op.drop_constraint(None, 'order_addons', type_='unique')
+    # ### end Alembic commands ###

alembic/versions/72e78a8a43cf_order_financials_and_invoice_link.py

@@ -0,0 +1,42 @@
+"""order financials and invoice link
+
+Revision ID: 72e78a8a43cf
+Revises: bc424721d767
+Create Date: 2026-08-02 13:55:18.013052
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '72e78a8a43cf'
+down_revision: Union[str, Sequence[str], None] = 'bc424721d767'
+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('invoices', sa.Column('order_id', sa.INTEGER(), nullable=True))
+    op.create_foreign_key(None, 'invoices', 'orders', ['order_id'], ['id'])
+    op.add_column('orders', sa.Column('total_amount', sa.FLOAT(), nullable=False))
+    op.add_column('orders', sa.Column('balance_amount', sa.FLOAT(), nullable=False))
+    op.add_column('orders', sa.Column('applies_at', sa.DateTime(timezone=True), nullable=True))
+    op.add_column('orders', sa.Column('applied_at', sa.DateTime(timezone=True), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('orders', 'applied_at')
+    op.drop_column('orders', 'applies_at')
+    op.drop_column('orders', 'balance_amount')
+    op.drop_column('orders', 'total_amount')
+    op.drop_constraint(None, 'invoices', type_='foreignkey')
+    op.drop_column('invoices', 'order_id')
+    # ### end Alembic commands ###

alembic/versions/b97ce5b7d663_subscriptions_infra.py

@@ -0,0 +1,72 @@
+"""subscriptions infra
+
+Revision ID: b97ce5b7d663
+Revises: 3d767875ec7d
+Create Date: 2026-08-02 12:38:25.357158
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b97ce5b7d663'
+down_revision: Union[str, Sequence[str], None] = '3d767875ec7d'
+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('orders',
+    sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('user_id', sa.INTEGER(), nullable=False),
+    sa.Column('devices', sa.INTEGER(), nullable=False),
+    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('id'),
+    sa.UniqueConstraint('user_id')
+    )
+    op.create_table('subscriptions',
+    sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('user_id', sa.INTEGER(), nullable=False),
+    sa.Column('devices', sa.INTEGER(), nullable=False),
+    sa.Column('status', sa.Enum('ACTIVE', 'EXPIRED', name='subscriptionstatus'), nullable=False),
+    sa.Column('expires_at', sa.DateTime(timezone=True), nullable=False),
+    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('id'),
+    sa.UniqueConstraint('user_id')
+    )
+    op.create_table('order_addons',
+    sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('order_id', sa.INTEGER(), nullable=False),
+    sa.Column('addon_id', sa.TEXT(), nullable=False),
+    sa.ForeignKeyConstraint(['addon_id'], ['addons.id'], ),
+    sa.ForeignKeyConstraint(['order_id'], ['orders.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('id')
+    )
+    op.create_table('subscription_addons',
+    sa.Column('id', sa.INTEGER(), autoincrement=True, nullable=False),
+    sa.Column('subscription_id', sa.INTEGER(), nullable=False),
+    sa.Column('addon_id', sa.TEXT(), nullable=False),
+    sa.ForeignKeyConstraint(['addon_id'], ['addons.id'], ),
+    sa.ForeignKeyConstraint(['subscription_id'], ['subscriptions.id'], ),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('subscription_addons')
+    op.drop_table('order_addons')
+    op.drop_table('subscriptions')
+    op.drop_table('orders')
+    # ### end Alembic commands ###

alembic/versions/bc424721d767_order_user_id_not_unique.py

@@ -0,0 +1,32 @@
+"""order.user_id -> NOT UNIQUE
+
+Revision ID: bc424721d767
+Revises: 0ea5625b4913
+Create Date: 2026-08-02 13:23:23.626697
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'bc424721d767'
+down_revision: Union[str, Sequence[str], None] = '0ea5625b4913'
+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.drop_constraint(op.f('orders_user_id_key'), 'orders', type_='unique')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_unique_constraint(op.f('orders_user_id_key'), 'orders', ['user_id'], postgresql_nulls_not_distinct=False)
+    # ### end Alembic commands ###

config.py

@@ -21,6 +21,8 @@ class Settings(BaseSettings):
     pally_shop_id: str = Field()
     pally_token: str = Field()
 
+    minimal_deposit: int = Field()
+
     @computed_field
     @property
     def db_url(self) -> str:

core/deps.py

@@ -6,6 +6,7 @@ from core.auth import jwt
 from db.session import get_db
 from external.pally import PallyClient
 from schemas.dto import AuthContext
+from services.subscriptions import sync_user_subscription
 
 
 async def get_auth_context(
@@ -18,7 +19,10 @@ async def get_auth_context(
 
     if auth.startswith("Bearer"):
         token = auth.removeprefix("Bearer ").strip()
-        return await jwt.authorize(token, session)
+        ctx = await jwt.authorize(token, session)
+        await sync_user_subscription(session, user=ctx.user)
+        await session.commit()
+        return ctx
 
 
 def get_pally_client() -> PallyClient:

db/models/__init__.py

@@ -1,8 +1,22 @@
 from .addons import Addon
 from .invoice import Invoice
+from .orders import Order, OrderAddon
 from .pricing import PricingConfig
 from .sessions import Session
+from .subscription_addons import SubscriptionAddon
+from .subscriptions import Subscription
 from .transactions import BalanceTransaction
 from .users import User
 
-__all__ = ["Addon", "BalanceTransaction", "Invoice", "PricingConfig", "Session", "User"]
+__all__ = [
+    "Addon",
+    "BalanceTransaction",
+    "Invoice",
+    "Order",
+    "OrderAddon",
+    "PricingConfig",
+    "Session",
+    "Subscription",
+    "SubscriptionAddon",
+    "User",
+]

db/models/invoice.py

@@ -7,7 +7,7 @@ from db.base import Base
 from schemas.invoices import InvoiceStatus
 
 if TYPE_CHECKING:
-    from db.models import User
+    from db.models import Order, User
 
 
 class Invoice(Base):
@@ -17,9 +17,11 @@ class Invoice(Base):
         INTEGER, autoincrement=True, unique=True, nullable=False, primary_key=True
     )
     creator_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
+    order_id: Mapped[int | None] = mapped_column(ForeignKey("orders.id"), nullable=True)
     amount: Mapped[float] = mapped_column(FLOAT, nullable=False)
     status: Mapped[InvoiceStatus] = mapped_column(
         Enum(InvoiceStatus, name="invoicestatus"), nullable=False, default=InvoiceStatus.ACTIVE
     )
 
     creator: Mapped["User"] = relationship("User", lazy="selectin")
+    order: Mapped["Order | None"] = relationship("Order", lazy="selectin")

db/models/orders.py

@@ -0,0 +1,53 @@
+import datetime
+from enum import StrEnum
+from typing import TYPE_CHECKING
+
+from sqlalchemy import FLOAT, INTEGER, DateTime, Enum, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from db.base import Base
+
+if TYPE_CHECKING:
+    from db.models import User
+
+
+class OrderStatus(StrEnum):
+    PENDING = "pending"
+    PAID = "paid"
+    DISCARDED = "discarded"
+
+
+class OrderAddon(Base):
+    __tablename__ = "order_addons"
+
+    id: Mapped[int] = mapped_column(
+        INTEGER, nullable=False, unique=True, autoincrement=True, primary_key=True
+    )
+    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), nullable=False)
+    addon_id: Mapped[str] = mapped_column(ForeignKey("addons.id"), nullable=False)
+
+    order: Mapped["Order"] = relationship("Order", back_populates="addons", lazy="selectin")
+
+
+class Order(Base):
+    __tablename__ = "orders"
+
+    id: Mapped[int] = mapped_column(
+        INTEGER, nullable=False, autoincrement=True, unique=True, primary_key=True
+    )
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
+
+    devices: Mapped[int] = mapped_column(INTEGER, nullable=False)
+    duration_days: Mapped[int] = mapped_column(INTEGER, nullable=False)
+    total_amount: Mapped[float] = mapped_column(FLOAT, nullable=False)
+    balance_amount: Mapped[float] = mapped_column(FLOAT, nullable=False, default=0)
+    status: Mapped[OrderStatus] = mapped_column(
+        Enum(OrderStatus, name="orderstatus"), nullable=False, default=OrderStatus.PENDING
+    )
+    applies_at: Mapped[datetime.datetime | None] = mapped_column(DateTime(True), nullable=True)
+    applied_at: Mapped[datetime.datetime | None] = mapped_column(DateTime(True), nullable=True)
+
+    user: Mapped["User"] = relationship("User", back_populates="orders", lazy="selectin")
+    addons: Mapped[list["OrderAddon"]] = relationship(
+        "OrderAddon", back_populates="order", lazy="selectin"
+    )

db/models/subscription_addons.py

@@ -0,0 +1,23 @@
+from typing import TYPE_CHECKING
+
+from sqlalchemy import INTEGER, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from db.base import Base
+
+if TYPE_CHECKING:
+    from db.models import Subscription
+
+
+class SubscriptionAddon(Base):
+    __tablename__ = "subscription_addons"
+
+    id: Mapped[int] = mapped_column(
+        INTEGER, nullable=False, unique=True, autoincrement=True, primary_key=True
+    )
+    subscription_id: Mapped[int] = mapped_column(ForeignKey("subscriptions.id"), nullable=False)
+    addon_id: Mapped[str] = mapped_column(ForeignKey("addons.id"), nullable=False)
+
+    subscription: Mapped["Subscription"] = relationship(
+        "Subscription", back_populates="addons", lazy="selectin"
+    )

db/models/subscriptions.py

@@ -0,0 +1,33 @@
+import datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import INTEGER, DateTime, Enum, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from db.base import Base
+from schemas.enums import SubscriptionStatus
+
+if TYPE_CHECKING:
+    from db.models import SubscriptionAddon, User
+
+
+class Subscription(Base):
+    __tablename__ = "subscriptions"
+
+    id: Mapped[int] = mapped_column(
+        INTEGER, nullable=False, autoincrement=True, unique=True, primary_key=True
+    )
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, unique=True)
+
+    devices: Mapped[int] = mapped_column(INTEGER, nullable=False)
+    status: Mapped[SubscriptionStatus] = mapped_column(
+        Enum(SubscriptionStatus, name="subscriptionstatus"),
+        nullable=False,
+        default=SubscriptionStatus.EXPIRED,
+    )
+    expires_at: Mapped[datetime.datetime] = mapped_column(DateTime(True), nullable=False)
+
+    user: Mapped["User"] = relationship("User", back_populates="subscription", lazy="selectin")
+    addons: Mapped[list["SubscriptionAddon"]] = relationship(
+        "SubscriptionAddon", back_populates="subscription", lazy="selectin"
+    )

db/models/users.py

@@ -6,7 +6,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
 from db.base import Base
 
 if TYPE_CHECKING:
-    from db.models.sessions import Session
+    from db.models import Order, Session, Subscription
 
 
 class User(Base):
@@ -22,6 +22,10 @@ class User(Base):
 
     balance: Mapped[float] = mapped_column(REAL, nullable=False, default=0)
 
+    orders: Mapped[list["Order"]] = relationship("Order", back_populates="user", lazy="selectin")
+    subscription: Mapped["Subscription"] = relationship(
+        "Subscription", back_populates="user", lazy="selectin"
+    )
     sessions: Mapped[list["Session"]] = relationship(back_populates="user", lazy="selectin")
     referal: Mapped["User | None"] = relationship(
         "User",

external/pally.py

@@ -224,7 +224,7 @@ class BillService(BaseService):
         super().__init__(session)
         self._payer_pays_commission = payer_pays_commission
 
-    async def create(
+    async def create(  # noqa: PLR0917
         self,
         amount: float,
         shop_id: str,

repositories/invoices.py

@@ -21,9 +21,12 @@ class InvoiceRepository:
 
         return list(r.scalars().all())
 
-    async def create(self, creator_id: int, amount: int | float, status: InvoiceStatus) -> Invoice:
+    async def create(
+        self, creator_id: int, order_id: int, amount: int | float, status: InvoiceStatus
+    ) -> Invoice:
         obj = Invoice(
             creator_id=creator_id,
+            order_id=order_id,
             amount=amount,
             status=status,
         )

repositories/orders.py

@@ -0,0 +1,74 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.models.orders import Order, OrderAddon, OrderStatus
+
+
+class OrderRepository:
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+
+    async def create(
+        self,
+        *,
+        user_id: int,
+        devices: int,
+        duration_days: int,
+        total_amount: float,
+        balance_amount: float,
+        addons: list[str] | None = None,
+        status: OrderStatus = OrderStatus.PENDING,
+    ) -> Order:
+        if not addons:
+            addons = []
+
+        order = Order(
+            user_id=user_id,
+            devices=devices,
+            duration_days=duration_days,
+            total_amount=total_amount,
+            balance_amount=balance_amount,
+            status=status,
+        )
+        self.session.add(order)
+        await self.session.flush()
+
+        for addon_id in addons:
+            addon = OrderAddon(order_id=order.id, addon_id=addon_id)
+            self.session.add(addon)
+
+        await self.session.commit()
+
+        return order
+
+    async def get_by_id(self, order_id: int) -> Order | None:
+        stmt = select(Order).where(Order.id == order_id)
+        r = await self.session.execute(stmt)
+
+        return r.scalar_one_or_none()
+
+    async def get_order_by_user_id(self, user_id: int) -> list[Order]:
+        stmt = select(Order).where(Order.user_id == user_id)
+        r = await self.session.execute(stmt)
+
+        return list(r.scalars().all())
+
+    async def get_active_by_user_id(self, user_id: int) -> list[Order]:
+        stmt = (
+            select(Order).where(Order.user_id == user_id).where(Order.status == OrderStatus.PENDING)
+        )
+        r = await self.session.execute(stmt)
+
+        return list(r.scalars().all())
+
+    async def get_paid_unapplied_by_user_id(self, user_id: int) -> list[Order]:
+        stmt = (
+            select(Order)
+            .where(Order.user_id == user_id)
+            .where(Order.status == OrderStatus.PAID)
+            .where(Order.applied_at.is_(None))
+            .order_by(Order.applies_at.asc(), Order.id.asc())
+        )
+        r = await self.session.execute(stmt)
+
+        return list(r.scalars().all())

routes/orders.py

@@ -1,21 +1,33 @@
+from datetime import UTC, datetime
+import math
+
 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 external.pally import PallyClient
 from repositories import AddonsRepository, PricingRepository
 from repositories.invoices import InvoiceRepository
+from repositories.orders import OrderRepository
+from schemas.checkout import CheckoutResponse
 from schemas.dto import AuthContext
-from schemas.invoices import InvoiceResponse, InvoiceStatus
+from schemas.invoices import InvoiceStatus
 from schemas.plans import OrderDetails
 from services.plans import calculate_price, get_pricing_model
+from services.subscriptions import (
+    apply_order_now,
+    deduct_order_balance,
+    queue_order_for_later,
+    should_apply_immediately,
+)
 
 router = APIRouter(prefix="/orders")
 
 
-@router.post("/checkout", response_model=InvoiceResponse, status_code=201)
+@router.post("/checkout", response_model=CheckoutResponse, status_code=201)
 async def checkout(
     order: OrderDetails,
     ctx: AuthContext = Depends(get_auth_context),
@@ -25,17 +37,72 @@ async def checkout(
     addons_repo = AddonsRepository(session)
     pricing_repo = PricingRepository(session)
     invoices_repo = InvoiceRepository(session)
+    orders_repo = OrderRepository(session)
 
     pricing = await get_pricing_model(addons_repo, pricing_repo)
-    price = await calculate_price(addons_repo=addons_repo, order=order, pricing=pricing)
+    price = math.ceil(await calculate_price(addons_repo=addons_repo, order=order, pricing=pricing))
+    bonus_covered = min(ctx.user.balance, price)
+    amount_to_pay = round(max(price - bonus_covered, 0), 2)
+    now = datetime.now(UTC)
 
-    invoice = await invoices_repo.create(
+    order_entry = await orders_repo.create(
-        creator_id=ctx.user.id, amount=price, status=InvoiceStatus.ACTIVE
+        user_id=ctx.user.id,
+        devices=order.devices,
+        duration_days=order.duration_days,
+        total_amount=price,
+        balance_amount=bonus_covered,
+        addons=order.addons,
     )
 
-    bill = await pally.bills.create(price, cfg.pally_shop_id, order_id=invoice.id)
+    if amount_to_pay > 0:
+        await invoices_repo.create(
+            creator_id=ctx.user.id,
+            order_id=order_entry.id,
+            amount=amount_to_pay,
+            status=InvoiceStatus.ACTIVE,
+        )
+        with open("log.txt", "w") as f:
+            f.write(
+                " ".join((str(price), str(amount_to_pay), cfg.pally_shop_id, str(order_entry.id)))
+            )
+        bill = await pally.bills.create(
+            float(amount_to_pay), cfg.pally_shop_id, order_id=str(order_entry.id)
+        )
+        payment_link = bill.link_page_url
+        if not payment_link:
+            raise HTTPException(500, detail="failed to create invoice")
+    else:
+        await deduct_order_balance(
+            session,
+            user=ctx.user,
+            order=order_entry,
+            description=f"order {order_entry.id} paid from balance",
+        )
+        order_entry.status = OrderStatus.PAID
+        if should_apply_immediately(
+            subscription=ctx.user.subscription,
+            order=order_entry,
+            pricing=pricing,
+            now=now,
+        ):
+            await apply_order_now(
+                session,
+                user=ctx.user,
+                order=order_entry,
+                pricing=pricing,
+                now=now,
+            )
+        else:
+            await queue_order_for_later(
+                order=order_entry, subscription=ctx.user.subscription, now=now
+            )
+        await session.commit()
+        payment_link = None
 
-    if not (bill.success and bill.link_page_url):
+    return CheckoutResponse(
-        raise HTTPException(500, detail="Failed to create an invoice.")
+        order_id=str(order_entry.id),
-
+        total_amount=price,
-    return InvoiceResponse(success=True, payment_link=bill.link_page_url, amount=float(price))
+        bonus_paid=bonus_covered,
+        amount_to_pay=amount_to_pay,
+        payment_link=payment_link,
+    )

routes/payments/pally.py

@@ -3,6 +3,7 @@ import hashlib
 import hmac
 import logging
 import math
+from datetime import UTC, datetime
 
 from fastapi import Depends, Form, HTTPException
 from fastapi.routing import APIRouter
@@ -10,11 +11,22 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import cfg
 from core.deps import get_db
-from db.models.transactions import BalanceTxType
+from db.models.orders import OrderStatus
+from db.models.transactions import BalanceTransaction, BalanceTxType
 from external.pally import BillStatus
+from repositories import AddonsRepository
 from repositories.invoices import InvoiceRepository
+from repositories.orders import OrderRepository
+from repositories.pricing import PricingRepository
 from repositories.users import UserRepository
 from schemas.invoices import InvoiceStatus
+from services.plans import get_pricing_model
+from services.subscriptions import (
+    apply_order_now,
+    deduct_order_balance,
+    queue_order_for_later,
+    should_apply_immediately,
+)
 
 router = APIRouter(prefix="/payments/pally")
 
@@ -22,7 +34,7 @@ logger = logging.getLogger(__name__)
 
 
 @router.post("/result")
-async def pally_callback(  # noqa: PLR0911
+async def pally_callback(  # noqa: PLR0911, PLR0912, PLR0915
     *,
     InvId: str = Form(...),
     OutSum: str = Form(...),
@@ -45,9 +57,13 @@ async def pally_callback(  # noqa: PLR0911
     ErrorMessage: str | None = Form(None),
     session: AsyncSession = Depends(get_db),
 ):
-    users_repo = UserRepository(session)
     invoice_repo = InvoiceRepository(session)
+    orders_repo = OrderRepository(session)
+    users_repo = UserRepository(session)
+    pricing_repo = PricingRepository(session)
     invoice_id_str = InvId
+    invoice_id: int | None = None
+    invoice_creator_id: int | None = None
 
     logger.info(
         "Pally webhook received - InvId: %s, OutSum: %s, Commission: %s, TrsId: %s, Status: %s, "
@@ -94,7 +110,7 @@ async def pally_callback(  # noqa: PLR0911
 
     logger.info("Processing successfully paid bill %s", TrsId)
 
-    # Validate bill ID (from InvId field, which contains the order_id from bill creation)
+    # Validate invoice ID
     if not invoice_id_str or not invoice_id_str.isdigit():
         logger.critical(
             "Invalid or non-numeric bill ID in InvId field for TrsId %s: '%s'",
@@ -110,6 +126,9 @@ async def pally_callback(  # noqa: PLR0911
         logger.critical("Bill %s not found in database for TrsId %s", invoice_id_str, TrsId)
         return "OK"
 
+    invoice_id = invoice.id
+    invoice_creator_id = invoice.creator_id
+
     # Check if already processed
     if invoice.status != InvoiceStatus.ACTIVE:
         logger.warning(
@@ -171,52 +190,120 @@ async def pally_callback(  # noqa: PLR0911
                 )
                 return "OK"
 
+        if invoice.order_id is None:
+            logger.critical("Invoice %s has no linked order for TrsId %s", invoice.id, TrsId)
+            return "OK"
+
+        order = await orders_repo.get_by_id(invoice.order_id)
+        if order is None:
+            logger.critical(
+                "Order not found for bill %s (TrsId: %s, user_id=%s, order_id=%s)",
+                invoice.id,
+                TrsId,
+                invoice.creator_id,
+                invoice.order_id,
+            )
+            return "OK"
+
+        if order.user_id != invoice.creator_id:
+            logger.critical(
+                "Order %s does not belong to invoice creator %s for TrsId %s",
+                order.id,
+                invoice.creator_id,
+                TrsId,
+            )
+            return "OK"
+
+        user = await users_repo.get_user_by_id(invoice.creator_id)
+        if user is None:
+            logger.critical("User %s not found for TrsId %s", invoice.creator_id, TrsId)
+            return "OK"
+
+        subscription = user.subscription
+        now = datetime.now(UTC)
+        pricing = await get_pricing_model(AddonsRepository(session), pricing_repo)
+
         logger.info(
-            "Processing payment: bill_id=%s, user_id=%s, amount=%s",
+            "Processing payment: bill_id=%s, order_id=%s, user_id=%s, amount=%s",
             invoice.id,
+            order.id,
             invoice.creator_id,
             amount,
         )
 
-        # Credit user balance
+        order.status = OrderStatus.PAID
-        await users_repo.increase_balance(
+        invoice.status = InvoiceStatus.PAID
-            invoice.creator_id,
+
-            amount=amount,
+        if order.balance_amount > 0:
-            tx_type=BalanceTxType.DEPOSIT,
+            await deduct_order_balance(
-            description=f"payment via PALLY (TrsId: {TrsId})",
+                session,
-        )
+                user=user,
+                order=order,
+                description=f"order {order.id} partial payment from balance",
+            )
+
+        if should_apply_immediately(
+            subscription=subscription,
+            order=order,
+            pricing=pricing,
+            now=now,
+        ):
+            await apply_order_now(session, user=user, order=order, pricing=pricing, now=now)
+        else:
+            if subscription is None:
+                logger.critical(
+                    "Cannot queue order %s without subscription for user %s", order.id, user.id
+                )
+                return "OK"
+            await queue_order_for_later(order=order, subscription=subscription, now=now)
 
         # Process referral bonus
-        user = invoice.creator
+        referal_id = user.referal_id
-        referal = user.referal
+        if referal_id is not None:
-        if referal is not None:
             referal_amount = math.floor(amount * (cfg.referal_bonus / 100))
-            await users_repo.increase_balance(
+            if referal_amount > 0:
-                referal,
+                referal_user = await session.get(type(user), referal_id)
-                referal_amount,
+                if referal_user is not None:
-                tx_type=BalanceTxType.REFERRAL_BONUS,
+                    session.add(
-                description=f"referral reward for user {invoice.creator_id} (TrsId: {TrsId})",
+                        BalanceTransaction(
-            )
+                            user_id=referal_id,
-            logger.info(
+                            amount=referal_amount,
-                "Referral bonus processed: referrer_id=%s, amount=%s", referal, referal_amount
+                            tx_type=BalanceTxType.REFERRAL_BONUS,
-            )
+                            balance_before=referal_user.balance,
+                            balance_after=referal_user.balance + referal_amount,
+                            description=f"referral reward for user {invoice.creator_id} (TrsId: {TrsId})",
+                        )
+                    )
+                    referal_user.balance += referal_amount
+
+                    logger.info(
+                        "Referral bonus processed: referrer_id=%s, amount=%s",
+                        referal_id,
+                        referal_amount,
+                    )
+                else:
+                    logger.warning(
+                        "Referrer %s not found for user_id=%s while processing TrsId %s",
+                        referal_id,
+                        invoice.creator_id,
+                        TrsId,
+                    )
+
+        await session.commit()
 
         logger.info("Payment processing completed successfully for TrsId %s", TrsId)
 
     except Exception as e:
+        await session.rollback()
         logger.exception(
             "CRITICAL ERROR processing payment for TrsId %s, bill_id %s, user_id %s: %s",
             TrsId,
-            invoice.id,
+            invoice_id,
-            invoice.creator_id,
+            invoice_creator_id,
             str(e),
         )
-        # Don't return early - still mark as success to prevent retries
+        raise HTTPException(500, detail="Payment processing failed.") from e
-        # The balance operation might have partially succeeded
 
-    # Update bill status to success
+    logger.info("Bill %s marked as PAID for TrsId %s", invoice_id, TrsId)
-    await invoice_repo.update_status_by_id(int(invoice.id), status=InvoiceStatus.PAID)
-
-    logger.info("Bill %s marked as SUCCESS for TrsId %s", invoice.id, TrsId)
 
     return "OK"

schemas/checkout.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel, Field, computed_field
+
+
+class CheckoutResponse(BaseModel):
+    order_id: str = Field()
+    total_amount: float = Field()
+    bonus_paid: float = Field()
+    amount_to_pay: float = Field()
+
+    payment_link: str | None = Field(None)
+
+    @property
+    @computed_field
+    def is_fully_paid(self):
+        return self.total_amount <= self.bonus_paid and self.amount_to_pay <= 0

schemas/enums.py

@@ -0,0 +1,6 @@
+from enum import StrEnum
+
+
+class SubscriptionStatus(StrEnum):
+    ACTIVE = "active"
+    EXPIRED = "expired"

services/plans.py

@@ -1,6 +1,7 @@
 from repositories.addons import AddonsRepository
 from repositories.pricing import PricingRepository
 from schemas.plans import AddonData, OrderDetails, PricingPlans
+from services.subscriptions import calculate_order_total
 
 
 async def get_pricing_model(addons_repo: AddonsRepository, pricing_repo: PricingRepository):
@@ -21,4 +22,9 @@ async def calculate_price(
     *, addons_repo: AddonsRepository, order: OrderDetails, pricing: PricingPlans
 ) -> float:
     addons = [await addons_repo.get_by_id(a) for a in order.addons]
-    return order.devices * pricing.device_price + sum([a.price for a in addons])
+    return calculate_order_total(
+        pricing,
+        order.devices,
+        [addon.id for addon in addons],
+        order.duration_days,
+    )

services/subscriptions.py

@@ -0,0 +1,185 @@
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import delete
+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 schemas.enums import SubscriptionStatus
+from schemas.plans import PricingPlans
+
+
+def calculate_plan_monthly_price(pricing: PricingPlans, devices: int, addon_ids: list[str]) -> float:
+    addon_prices = {addon.id: addon.price for addon in pricing.addons}
+    return devices * pricing.device_price + sum(addon_prices[addon_id] for addon_id in addon_ids)
+
+
+def calculate_order_total(pricing: PricingPlans, devices: int, addon_ids: list[str], duration_days: int) -> float:
+    monthly_price = calculate_plan_monthly_price(pricing, devices, addon_ids)
+    return monthly_price * duration_days / 30
+
+
+def get_subscription_addon_ids(subscription: Subscription | None) -> list[str]:
+    if subscription is None:
+        return []
+
+    return [addon.addon_id for addon in subscription.addons]
+
+
+def should_apply_immediately(
+    *,
+    subscription: Subscription | None,
+    order: Order,
+    pricing: PricingPlans,
+    now: datetime,
+) -> bool:
+    if subscription is None or subscription.expires_at <= now:
+        return True
+
+    current_addons = set(get_subscription_addon_ids(subscription))
+    new_addons = {addon.addon_id for addon in order.addons}
+
+    if order.devices == subscription.devices and new_addons == current_addons:
+        return True
+
+    current_monthly_price = calculate_plan_monthly_price(
+        pricing, subscription.devices, list(current_addons)
+    )
+    new_monthly_price = calculate_plan_monthly_price(pricing, order.devices, list(new_addons))
+
+    return (
+        order.devices >= subscription.devices
+        and new_addons.issuperset(current_addons)
+        and new_monthly_price >= current_monthly_price
+    )
+
+
+async def replace_subscription_addons(
+    session: AsyncSession, subscription_id: int, addon_ids: list[str]
+) -> None:
+    await session.execute(
+        delete(SubscriptionAddon).where(SubscriptionAddon.subscription_id == subscription_id)
+    )
+
+    for addon_id in addon_ids:
+        session.add(SubscriptionAddon(subscription_id=subscription_id, addon_id=addon_id))
+
+
+async def ensure_subscription(user: User, session: AsyncSession, starts_at: datetime) -> Subscription:
+    subscription = user.subscription
+    if subscription is not None:
+        return subscription
+
+    subscription = Subscription(
+        user_id=user.id,
+        devices=0,
+        status=SubscriptionStatus.EXPIRED,
+        expires_at=starts_at,
+    )
+    session.add(subscription)
+    await session.flush()
+    user.subscription = subscription
+    return subscription
+
+
+async def deduct_order_balance(
+    session: AsyncSession,
+    *,
+    user: User,
+    order: Order,
+    description: str,
+) -> None:
+    if order.balance_amount <= 0:
+        return
+
+    session.add(
+        BalanceTransaction(
+            user_id=user.id,
+            amount=-order.balance_amount,
+            tx_type=BalanceTxType.PURCHASE,
+            balance_before=user.balance,
+            balance_after=user.balance - order.balance_amount,
+            description=description,
+        )
+    )
+    user.balance -= order.balance_amount
+
+
+async def apply_order_now(
+    session: AsyncSession,
+    *,
+    user: User,
+    order: Order,
+    pricing: PricingPlans,
+    now: datetime,
+) -> None:
+    subscription = await ensure_subscription(user, session, now)
+    addon_ids = [addon.addon_id for addon in order.addons]
+    current_addons = [] if subscription.devices == 0 else get_subscription_addon_ids(subscription)
+
+    if subscription.status == SubscriptionStatus.ACTIVE and subscription.expires_at > now:
+        if order.devices == subscription.devices and set(addon_ids) == set(current_addons):
+            subscription.expires_at += timedelta(days=order.duration_days)
+        else:
+            current_monthly_price = calculate_plan_monthly_price(
+                pricing, subscription.devices, current_addons
+            )
+            next_monthly_price = calculate_plan_monthly_price(pricing, order.devices, addon_ids)
+            remaining_seconds = (subscription.expires_at - now).total_seconds()
+            remaining_days = max(remaining_seconds / 86400, 0)
+            remaining_credit = current_monthly_price * remaining_days / 30
+            purchased_value = calculate_order_total(
+                pricing, order.devices, addon_ids, order.duration_days
+            )
+            total_days = ((remaining_credit + purchased_value) / next_monthly_price) * 30
+            subscription.expires_at = now + timedelta(days=total_days)
+    else:
+        subscription.expires_at = now + timedelta(days=order.duration_days)
+
+    subscription.devices = order.devices
+    subscription.status = SubscriptionStatus.ACTIVE
+    await replace_subscription_addons(session, subscription.id, addon_ids)
+    order.applies_at = now
+    order.applied_at = now
+
+
+async def queue_order_for_later(
+    *, order: Order, subscription: Subscription, now: datetime
+) -> None:
+    order.applies_at = max(subscription.expires_at, now)
+
+
+async def sync_user_subscription(
+    session: AsyncSession,
+    *,
+    user: User,
+) -> None:
+    now = datetime.now(UTC)
+    subscription = user.subscription
+
+    if subscription is not None and subscription.expires_at <= now:
+        due_orders = [
+            order
+            for order in user.orders
+            if order.status == OrderStatus.PAID
+            and order.applied_at is None
+            and order.applies_at is not None
+            and order.applies_at <= now
+        ]
+        due_orders.sort(key=lambda order: (order.applies_at, order.id))
+
+        if due_orders:
+            for order in due_orders:
+                base_time = max(subscription.expires_at, order.applies_at)
+                subscription.devices = order.devices
+                subscription.status = SubscriptionStatus.ACTIVE
+                subscription.expires_at = base_time + timedelta(days=order.duration_days)
+                await replace_subscription_addons(
+                    session,
+                    subscription.id,
+                    [addon.addon_id for addon in order.addons],
+                )
+                order.applied_at = now
+        else:
+            subscription.status = SubscriptionStatus.EXPIRED