"""Premium subscription service layer. Encapsulates the manual activation workflow (Phase 1): students submit a :class:`~app.models.subscription.PremiumRequest`, a Super Admin approves or rejects it, and approval creates an active :class:`~app.models.subscription.Subscription`. Premium *entitlement* is resolved live from the subscription table (``status == ACTIVE`` and ``end_date`` in the future) so an expired subscription loses its benefits automatically without a background job. Nothing here unlocks any premium feature yet — that is a later phase; this module only owns the subscription lifecycle and the single ``is_user_premium`` entitlement check. """ from __future__ import annotations from datetime import datetime, timedelta from math import ceil from flask import current_app from app.extensions import db from app.models import ( NotificationType, PremiumRequest, PremiumRequestStatus, Subscription, SubscriptionStatus, User, ) from app.models.base import utcnow # A direct premium activation lasts this many days. Defined once so the duration # can be changed in a single place (mirrors ``FEATURE_DURATION_DAYS``). PREMIUM_DURATION_DAYS = 30 class PremiumError(Exception): """Base error for the premium workflow.""" class DuplicateRequestError(PremiumError): """Raised when a user already has a pending premium request.""" class InvalidPlanError(PremiumError): """Raised when an unknown plan name is supplied.""" class RequestNotPendingError(PremiumError): """Raised when reviewing a request that is not in the PENDING state.""" # Plan catalog. Prices are DISPLAY-ONLY — Student Bazaar does not process # payments, so nothing is charged. ``duration_days`` drives the subscription # end date on approval. PLAN_CATALOG = { "monthly": { "label": "Monthly", "price": 99, "duration_days": 30, "tagline": "Billed monthly", }, "yearly": { "label": "Yearly", "price": 999, "duration_days": 365, "tagline": "Best value — save over 15%", }, } # Accepted out-of-band payment methods (value, label). A cash payment needs no # transaction reference; the others do (enforced by the form). PAYMENT_METHODS = [ ("upi", "UPI"), ("bank_transfer", "Bank Transfer"), ("cash", "Cash"), ] CURRENCY_SYMBOL = "₹" def plan_choices() -> list[tuple[str, str]]: """Return ``(value, label)`` plan options for select fields (admin-configured).""" from app.premium.config import plan_choices as config_plan_choices return config_plan_choices() def is_valid_plan(plan_name: str) -> bool: """True when ``plan_name`` matches a configured plan (case-insensitive).""" from app.premium.config import get_plan_by_name return get_plan_by_name(plan_name) is not None def _plan_duration_days(plan_name: str) -> int | None: """Configured duration for a plan (falls back to the legacy catalog).""" from app.premium.config import get_plan_by_name plan = get_plan_by_name(plan_name) if plan is not None: return plan.duration_days legacy = PLAN_CATALOG.get((plan_name or "").lower()) return legacy["duration_days"] if legacy else None def _notify_premium(user: User, title: str, message: str) -> None: """Send a Premium notification (reuses the shared notification service). Never raises into the caller — a delivery failure must not undo the action.""" if user is None: return from app.notifications.services import create_notification try: create_notification(user, NotificationType.PREMIUM, title, message) except Exception: # pragma: no cover - notifications must not break the workflow db.session.rollback() current_app.logger.exception("Failed to send premium notification to user %s", getattr(user, "id", None)) # --- Entitlement ----------------------------------------------------------- def get_active_subscription(user: User) -> Subscription | None: """Return the user's active, non-expired subscription, or ``None``. This is the single source of truth for premium status. A subscription counts as active only while ``status == ACTIVE`` and ``end_date`` is in the future, so expiry takes effect automatically. """ return db.session.scalar( db.select(Subscription) .where( Subscription.user_id == user.id, Subscription.status == SubscriptionStatus.ACTIVE, Subscription.end_date > utcnow(), ) .order_by(Subscription.end_date.desc()) ) def is_user_premium(user: User) -> bool: """Return ``True`` when the user currently has premium entitlement.""" if user is None or not getattr(user, "id", None): return False return get_active_subscription(user) is not None # --- Subscription expiry (Phase 3) ----------------------------------------- # # Premium already expires *live* — ``get_active_subscription`` requires # ``end_date > now`` — so entitlement is always correct regardless of stored # status. The helpers below give that behaviour the same shape as the Featured # Listing expiry (a duration constant, a "currently premium" check, a # days-remaining helper, direct activate/expire, and a bulk cleanup) without a # background job and without a second source of truth on the ``users`` table. def is_user_currently_premium(user: User) -> bool: """Canonical live premium check: an ACTIVE subscription whose ``end_date`` is still in the future. Never relies on a stored flag; one indexed query.""" return is_user_premium(user) def premium_expiry(user: User) -> datetime | None: """The end date of the user's active subscription, or ``None``.""" subscription = get_active_subscription(user) return subscription.end_date if subscription else None def premium_days_remaining(user: User) -> int: """Whole days remaining on the active subscription; ``0`` when none/expired. Rounded up so a fresh activation reads as the full duration. All date maths for premium lives here — templates never compute it. """ subscription = get_active_subscription(user) if subscription is None: return 0 seconds = (subscription.end_date - utcnow()).total_seconds() return max(int(ceil(seconds / 86400)), 0) def activate_premium( user: User, *, duration_days: int = PREMIUM_DURATION_DAYS, plan_name: str = "monthly", activated_by: int | None = None, ) -> Subscription: """Activate (or renew) premium for ``user`` with a fresh window. Resets the timer: any existing active subscription is cancelled and a new ACTIVE subscription is created for ``duration_days``, preserving the single-active invariant used by the admin approval flow. History is kept. """ now = utcnow() existing = get_active_subscription(user) if existing is not None: existing.status = SubscriptionStatus.CANCELLED subscription = Subscription( user_id=user.id, plan_name=plan_name, status=SubscriptionStatus.ACTIVE, start_date=now, end_date=now + timedelta(days=duration_days), activated_by=activated_by, ) db.session.add(subscription) db.session.commit() return subscription def activate_premium_from_payment(user: User, plan_name: str) -> Subscription: """Activate premium after a verified online payment (gateway-agnostic). Reuses :func:`activate_premium` for the subscription lifecycle (single-active invariant + history) and the *same* "Premium Activated" notification and activity-feed entry as the manual approval flow, so entitlement and messaging are identical no matter how premium was purchased. The payment layer calls this after signature/webhook verification; it never touches gateway details. """ duration_days = _plan_duration_days(plan_name) or PREMIUM_DURATION_DAYS subscription = activate_premium(user, duration_days=duration_days, plan_name=plan_name) _notify_premium(user, "Premium Activated", "Your payment was verified and Premium is now active on your account.") _publish_premium_activity(user, "premium_approved", "Premium activated", "Your Premium membership is now active. Enjoy the benefits!") return subscription def expire_premium(user: User) -> int: """Canonical premium expiry: mark the user's ACTIVE subscription(s) EXPIRED. Preserves history (the row is retained as EXPIRED, never deleted). Returns how many subscriptions were expired. """ subscriptions = db.session.scalars( db.select(Subscription).where( Subscription.user_id == user.id, Subscription.status == SubscriptionStatus.ACTIVE, ) ).all() for subscription in subscriptions: subscription.status = SubscriptionStatus.EXPIRED if subscriptions: db.session.commit() return len(subscriptions) def cleanup_expired_premium_users() -> int: """Flip every lapsed ACTIVE subscription to EXPIRED in one bulk UPDATE. O(1), no loops / N+1 — mirrors ``cleanup_expired_featured_listings``. Run transparently at premium-related entry points; entitlement is already correct via the live check, so this only keeps the stored status accurate (and powers "Premium Expired" displays). Returns the number expired. """ result = db.session.execute( db.update(Subscription) .where( Subscription.status == SubscriptionStatus.ACTIVE, Subscription.end_date <= utcnow(), ) .values(status=SubscriptionStatus.EXPIRED) ) if result.rowcount: db.session.commit() return result.rowcount or 0 # --- Requests -------------------------------------------------------------- def get_pending_request(user: User) -> PremiumRequest | None: """Return the user's open (pending) request, if any.""" return db.session.scalar( db.select(PremiumRequest).where( PremiumRequest.user_id == user.id, PremiumRequest.status == PremiumRequestStatus.PENDING, ) ) def save_premium_receipt(file) -> str | None: """Validate and persist an optional payment-receipt screenshot; return the stored (opaque) filename, or ``None`` when no file was supplied. Raises ``ValueError`` for a non-image upload.""" import io import uuid from PIL import Image from werkzeug.utils import secure_filename if file is None or not getattr(file, "filename", ""): return None ext = (file.filename.rsplit(".", 1)[-1] if "." in file.filename else "").lower() data = file.read() try: Image.open(io.BytesIO(data)).verify() except Exception as exc: # noqa: BLE001 - any Pillow failure means "not an image" raise ValueError("The receipt must be an image file.") from exc safe_ext = secure_filename(f"x.{ext or 'png'}").rsplit(".", 1)[-1] or "png" stored_name = f"{uuid.uuid4().hex}.{safe_ext}" with open(current_app.config["PREMIUM_RECEIPT_FOLDER"] / stored_name, "wb") as out: out.write(data) return stored_name def create_premium_request( user: User, plan_name: str, payment_method: str, transaction_reference: str | None = None, remarks: str | None = None, receipt_path: str | None = None, ) -> PremiumRequest: """Create a PENDING premium request for a user. Guards against submitting a second request while one is already pending. """ if not is_valid_plan(plan_name): raise InvalidPlanError(f"Unknown plan: {plan_name!r}.") if get_pending_request(user) is not None: raise DuplicateRequestError( "You already have a premium request awaiting review." ) request = PremiumRequest( user_id=user.id, plan_name=plan_name, payment_method=payment_method, transaction_reference=(transaction_reference or None), receipt_path=(receipt_path or None), remarks=(remarks or None), status=PremiumRequestStatus.PENDING, requested_at=utcnow(), ) db.session.add(request) db.session.commit() _notify_premium(user, "Premium Request Submitted", "We received your Premium request and it's awaiting verification.") return request def approve_premium_request(request: PremiumRequest, admin: User) -> Subscription: """Approve a request and activate an ACTIVE subscription. Marks the request APPROVED, records the reviewing admin, and creates a subscription whose duration matches the requested plan. To preserve the "one active subscription per student" rule, any existing active subscription is cancelled first (history is retained). """ if request.status != PremiumRequestStatus.PENDING: raise RequestNotPendingError("This request has already been reviewed.") duration_days = _plan_duration_days(request.plan_name) if duration_days is None: raise InvalidPlanError(f"Unknown plan: {request.plan_name!r}.") now = utcnow() # Preserve the single-active-subscription invariant. existing = get_active_subscription(request.user) if existing is not None: existing.status = SubscriptionStatus.CANCELLED subscription = Subscription( user_id=request.user_id, plan_name=request.plan_name, status=SubscriptionStatus.ACTIVE, start_date=now, end_date=now + timedelta(days=duration_days), activated_by=admin.id, ) db.session.add(subscription) request.status = PremiumRequestStatus.APPROVED request.reviewed_at = now request.reviewed_by = admin.id db.session.commit() _notify_premium(request.user, "Premium Activated", "Your payment was verified and Premium is now active on your account.") _publish_premium_activity(request.user, "premium_approved", "Premium activated", "Your Premium membership is now active. Enjoy the benefits!") return subscription def _publish_premium_activity(user, event_type, title, message): """Route premium events through the single activity API (feed + separate notif).""" from app.activity.services import publish_activity publish_activity(user, event_type, title, message=message, related_entity_type=None, related_entity_id=None) def reject_premium_request(request: PremiumRequest, admin: User, reason: str | None = None) -> PremiumRequest: """Reject a request (with an optional reason). No subscription is created; the student may submit a fresh request afterwards.""" if request.status != PremiumRequestStatus.PENDING: raise RequestNotPendingError("This request has already been reviewed.") request.status = PremiumRequestStatus.REJECTED request.reviewed_at = utcnow() request.reviewed_by = admin.id request.review_note = (reason or "").strip() or None db.session.commit() message = "Your Premium request was not approved." if request.review_note: message += f" Reason: {request.review_note}" message += " You can submit a new request once resolved." _notify_premium(request.user, "Premium Request Update", message) return request def request_more_info(request: PremiumRequest, admin: User, message: str) -> PremiumRequest: """Ask the student for more information. Moves the request to MORE_INFO (so it leaves the pending queue) and lets them submit an updated request.""" if request.status != PremiumRequestStatus.PENDING: raise RequestNotPendingError("This request has already been reviewed.") request.status = PremiumRequestStatus.MORE_INFO request.reviewed_at = utcnow() request.reviewed_by = admin.id request.review_note = (message or "").strip() or None db.session.commit() body = "We need a little more information to verify your Premium payment." if request.review_note: body += f" {request.review_note}" body += " Please submit an updated request." _notify_premium(request.user, "More Information Needed", body) return request def get_latest_request(user: User) -> PremiumRequest | None: """The user's most recent request of any status (for the status page).""" return db.session.scalar( db.select(PremiumRequest) .where(PremiumRequest.user_id == user.id) .order_by(PremiumRequest.requested_at.desc(), PremiumRequest.id.desc()) ) # --- Admin queries --------------------------------------------------------- def get_request_or_404(public_id: str) -> PremiumRequest: """Fetch a request by public id or abort with 404.""" return db.first_or_404( db.select(PremiumRequest).where(PremiumRequest.public_id == public_id) ) def list_premium_requests(status: str | None = None): """Return all premium requests (optionally filtered by status), newest first.""" query = db.select(PremiumRequest).order_by(PremiumRequest.requested_at.desc()) if status: try: query = query.where(PremiumRequest.status == PremiumRequestStatus(status)) except ValueError: pass # Unknown status filter — ignore and return all. return db.session.scalars(query).all() def get_user_requests(user: User): """Return a user's own requests, newest first (for their premium page).""" return db.session.scalars( db.select(PremiumRequest) .where(PremiumRequest.user_id == user.id) .order_by(PremiumRequest.requested_at.desc()) ).all() def get_premium_seller_ids(items) -> set[int]: """Resolve which sellers on a listing page currently have active premium. Accepts any iterable of listings (objects exposing ``seller_id``) or plain integer user ids, and returns the subset that have an active, non-expired subscription — using **one** query for the whole page instead of a lookup per card (avoids the badge N+1 on listing grids). Uses the same entitlement definition as :func:`get_active_subscription` (``status == ACTIVE`` and ``end_date`` in the future). De-duplicates sellers and returns an empty set for an empty page or when there are no premium sellers. """ seller_ids: set[int] = set() for item in items or (): sid = getattr(item, "seller_id", item) if isinstance(sid, int): seller_ids.add(sid) if not seller_ids: return set() rows = db.session.scalars( db.select(Subscription.user_id).where( Subscription.user_id.in_(seller_ids), Subscription.status == SubscriptionStatus.ACTIVE, Subscription.end_date > utcnow(), ) ).all() return set(rows) def get_latest_subscription(user: User) -> Subscription | None: """Return the user's most recent subscription regardless of status. Read-only helper used by the profile status card to distinguish a user who never subscribed ("Free") from one whose subscription has lapsed ("Expired"). """ return db.session.scalar( db.select(Subscription) .where(Subscription.user_id == user.id) .order_by(Subscription.start_date.desc()) )