feat(sub): handling upgrade/downgrade

rw integration and /me endpoints are left

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 ###

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/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

@@ -1,7 +1,8 @@
+import datetime
 from enum import StrEnum
 from typing import TYPE_CHECKING
 
-from sqlalchemy import INTEGER, Enum, ForeignKey
+from sqlalchemy import FLOAT, INTEGER, DateTime, Enum, ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from db.base import Base
@@ -38,9 +39,13 @@ class Order(Base):
 
     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(

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

@@ -10,9 +10,12 @@ class OrderRepository:
 
     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:
@@ -23,6 +26,8 @@ class OrderRepository:
             user_id=user_id,
             devices=devices,
             duration_days=duration_days,
+            total_amount=total_amount,
+            balance_amount=balance_amount,
             status=status,
         )
         self.session.add(order)
@@ -36,6 +41,12 @@ class OrderRepository:
 
         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)
@@ -49,3 +60,15 @@ class OrderRepository:
         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,8 +1,12 @@
+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
@@ -13,6 +17,12 @@ from schemas.dto import AuthContext
 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")
 
@@ -29,32 +39,68 @@ async def checkout(
     invoices_repo = InvoiceRepository(session)
     orders_repo = OrderRepository(session)
 
+    pricing = await get_pricing_model(addons_repo, pricing_repo)
+    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)
+
     order_entry = await orders_repo.create(
-        ctx.user.id,
+        user_id=ctx.user.id,
-        order.devices,
+        devices=order.devices,
-        order.duration_days,
+        duration_days=order.duration_days,
+        total_amount=price,
+        balance_amount=bonus_covered,
         addons=order.addons,
     )
 
-    pricing = await get_pricing_model(addons_repo, pricing_repo)
-    price = await calculate_price(addons_repo=addons_repo, order=order, pricing=pricing)
-    bonus_covered = ctx.user.balance
-    amount_to_pay = max(price - bonus_covered, cfg.minimal_deposit) if bonus_covered < price else 0
-
     if amount_to_pay > 0:
-        invoice = await invoices_repo.create(
+        await invoices_repo.create(
-            creator_id=ctx.user.id, amount=amount_to_pay, status=InvoiceStatus.ACTIVE
+            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)
         )
-
-        bill = await pally.bills.create(amount_to_pay, cfg.pally_shop_id, order_id=invoice.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
 
     return CheckoutResponse(
-        order_id=order_entry.id,
+        order_id=str(order_entry.id),
         total_amount=price,
         bonus_paid=bonus_covered,
         amount_to_pay=amount_to_pay,

routes/payments/pally.py

@@ -1,26 +1,32 @@
 # ruff: noqa: N803
-from datetime import UTC, datetime, timedelta
 import hashlib
 import hmac
 import logging
 import math
+from datetime import UTC, datetime
 
 from fastapi import Depends, Form, HTTPException
 from fastapi.routing import APIRouter
-from sqlalchemy import delete
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import cfg
 from core.deps import get_db
-from db.models import Subscription, SubscriptionAddon
 from db.models.orders import OrderStatus
-from db.models.transactions import BalanceTransaction
+from db.models.transactions import BalanceTransaction, BalanceTxType
-from db.models.transactions import BalanceTxType
 from external.pally import BillStatus
+from repositories import AddonsRepository
 from repositories.invoices import InvoiceRepository
 from repositories.orders import OrderRepository
-from schemas.enums import SubscriptionStatus
+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")
 
@@ -28,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(...),
@@ -53,7 +59,11 @@ async def pally_callback(  # noqa: PLR0911
 ):
     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, "
@@ -100,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'",
@@ -116,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(
@@ -177,20 +190,38 @@ async def pally_callback(  # noqa: PLR0911
                 )
                 return "OK"
 
-        active_orders = await orders_repo.get_active_by_user_id(invoice.creator_id)
+        if invoice.order_id is None:
-        if not active_orders:
+            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(
-                "Active order not found for bill %s (TrsId: %s, user_id=%s)",
+                "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"
 
-        order = active_orders[0]
+        if order.user_id != invoice.creator_id:
-        user = invoice.creator
+            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, order_id=%s, user_id=%s, amount=%s",
@@ -203,32 +234,28 @@ async def pally_callback(  # noqa: PLR0911
         order.status = OrderStatus.PAID
         invoice.status = InvoiceStatus.PAID
 
-        if subscription is None:
+        if order.balance_amount > 0:
-            subscription = Subscription(
+            await deduct_order_balance(
-                user_id=invoice.creator_id,
+                session,
-                devices=order.devices,
+                user=user,
-                status=SubscriptionStatus.ACTIVE,
+                order=order,
-                expires_at=now + timedelta(days=order.duration_days),
+                description=f"order {order.id} partial payment from balance",
-            )
-            session.add(subscription)
-            await session.flush()
-        else:
-            logger.warning(
-                "Subscription already exists for user_id=%s, updating existing subscription %s",
-                invoice.creator_id,
-                subscription.id,
-            )
-            subscription.devices = order.devices
-            subscription.status = SubscriptionStatus.ACTIVE
-            subscription.expires_at = max(subscription.expires_at, now) + timedelta(
-                days=order.duration_days
-            )
-            await session.execute(
-                delete(SubscriptionAddon).where(SubscriptionAddon.subscription_id == subscription.id)
             )
 
-        for addon in order.addons:
+        if should_apply_immediately(
-            session.add(SubscriptionAddon(subscription_id=subscription.id, addon_id=addon.addon_id))
+            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
         referal_id = user.referal_id
@@ -271,12 +298,12 @@ async def pally_callback(  # noqa: PLR0911
         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),
         )
         raise HTTPException(500, detail="Payment processing failed.") from e
 
-    logger.info("Bill %s marked as PAID for TrsId %s", invoice.id, TrsId)
+    logger.info("Bill %s marked as PAID for TrsId %s", invoice_id, TrsId)
 
     return "OK"

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