From 97a4e819d6e9a2b46aae4b0e0e89c0d425a4f709 Mon Sep 17 00:00:00 2001 From: hexdev Date: Sun, 2 Aug 2026 14:16:57 +0700 Subject: [PATCH] feat(sub): handling upgrade/downgrade rw integration and /me endpoints are left --- ...a43cf_order_financials_and_invoice_link.py | 42 ++++ core/deps.py | 6 +- db/models/invoice.py | 4 +- db/models/orders.py | 7 +- repositories/invoices.py | 5 +- repositories/orders.py | 23 +++ routes/orders.py | 72 +++++-- routes/payments/pally.py | 107 ++++++---- services/plans.py | 8 +- services/subscriptions.py | 185 ++++++++++++++++++ 10 files changed, 401 insertions(+), 58 deletions(-) create mode 100644 alembic/versions/72e78a8a43cf_order_financials_and_invoice_link.py create mode 100644 services/subscriptions.py diff --git a/alembic/versions/72e78a8a43cf_order_financials_and_invoice_link.py b/alembic/versions/72e78a8a43cf_order_financials_and_invoice_link.py new file mode 100644 index 0000000..6b76437 --- /dev/null +++ b/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 ### diff --git a/core/deps.py b/core/deps.py index 97cdee4..e3239c7 100644 --- a/core/deps.py +++ b/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: diff --git a/db/models/invoice.py b/db/models/invoice.py index 4033a90..40e338e 100644 --- a/db/models/invoice.py +++ b/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") diff --git a/db/models/orders.py b/db/models/orders.py index de493c0..2b14b3a 100644 --- a/db/models/orders.py +++ b/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( diff --git a/repositories/invoices.py b/repositories/invoices.py index fee85af..2873484 100644 --- a/repositories/invoices.py +++ b/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, ) diff --git a/repositories/orders.py b/repositories/orders.py index 54f8883..11f2f02 100644 --- a/repositories/orders.py +++ b/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()) diff --git a/routes/orders.py b/routes/orders.py index 2937926..90046d2 100644 --- a/routes/orders.py +++ b/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, - order.devices, - order.duration_days, + user_id=ctx.user.id, + devices=order.devices, + 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( - creator_id=ctx.user.id, amount=amount_to_pay, status=InvoiceStatus.ACTIVE + 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) ) - - 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, diff --git a/routes/payments/pally.py b/routes/payments/pally.py index 90be1a1..cdba2de 100644 --- a/routes/payments/pally.py +++ b/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 BalanceTxType +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 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 not active_orders: + 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( - "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] - user = invoice.creator + 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, 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: - subscription = Subscription( - user_id=invoice.creator_id, - devices=order.devices, - status=SubscriptionStatus.ACTIVE, - expires_at=now + timedelta(days=order.duration_days), - ) - 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) + if order.balance_amount > 0: + await deduct_order_balance( + session, + user=user, + order=order, + description=f"order {order.id} partial payment from balance", ) - for addon in order.addons: - session.add(SubscriptionAddon(subscription_id=subscription.id, addon_id=addon.addon_id)) + 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 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.creator_id, + invoice_id, + invoice_creator_id, str(e), ) raise HTTPException(500, detail="Payment processing failed.") from e - logger.info("Bill %s marked as PAID for TrsId %s", invoice.id, TrsId) + logger.info("Bill %s marked as PAID for TrsId %s", invoice_id, TrsId) return "OK" diff --git a/services/plans.py b/services/plans.py index ed778ee..cd97cc2 100644 --- a/services/plans.py +++ b/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, + ) diff --git a/services/subscriptions.py b/services/subscriptions.py new file mode 100644 index 0000000..12a324c --- /dev/null +++ b/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