From a83ea716e41888bd0b4714f50933750a40012374 Mon Sep 17 00:00:00 2001 From: unandhakumar2611 Date: Mon, 17 Aug 2026 12:34:09 +0530 Subject: [PATCH 1/3] subscription added --- backend/app/celery_app.py | 4 + backend/app/main.py | 3 +- backend/app/models/__init__.py | 2 + backend/app/models/subscription.py | 7 +- backend/app/models/usage.py | 20 ++ backend/app/routers/admin.py | 34 +++ backend/app/routers/ai.py | 33 ++- backend/app/routers/billing.py | 186 ++++++++-------- backend/app/routers/jobs.py | 24 +- backend/app/routers/webhooks.py | 50 +---- backend/app/schemas/admin.py | 51 +++++ backend/app/schemas/billing.py | 22 +- backend/app/schemas/usage.py | 27 +++ backend/app/services/admin.py | 142 ++++++++++++ backend/app/services/ai.py | 38 ---- backend/app/services/subscription.py | 140 ++++++++++++ backend/app/services/usage.py | 150 +++++++++++++ backend/app/services/webhook.py | 210 ++++++++++++++++++ backend/app/tasks/usage_reset.py | 71 ++++++ backend/scripts/promote_admin.py | 56 +++++ database/migrations/001_init.sql | 54 ++++- frontend/src/App.tsx | 13 ++ frontend/src/api/admin.ts | 61 +++++ frontend/src/api/billing.ts | 43 ++++ frontend/src/components/app-shell.tsx | 23 +- frontend/src/components/job-detail-dialog.tsx | 28 +-- frontend/src/components/paywall-dialog.tsx | 55 ++++- frontend/src/pages/admin/overview.tsx | 81 +++++++ frontend/src/pages/admin/subscriptions.tsx | 75 +++++++ frontend/src/pages/billing.tsx | 165 ++++++++++++-- frontend/src/pages/builder.tsx | 20 +- frontend/src/pages/generate.tsx | 7 +- frontend/src/pages/jobs.tsx | 33 ++- 33 files changed, 1684 insertions(+), 244 deletions(-) create mode 100644 backend/app/models/usage.py create mode 100644 backend/app/routers/admin.py create mode 100644 backend/app/schemas/admin.py create mode 100644 backend/app/schemas/usage.py create mode 100644 backend/app/services/admin.py create mode 100644 backend/app/services/subscription.py create mode 100644 backend/app/services/usage.py create mode 100644 backend/app/services/webhook.py create mode 100644 backend/app/tasks/usage_reset.py create mode 100644 backend/scripts/promote_admin.py create mode 100644 frontend/src/api/admin.ts create mode 100644 frontend/src/pages/admin/overview.tsx create mode 100644 frontend/src/pages/admin/subscriptions.tsx diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 1d78a5b..fd383c5 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -29,6 +29,10 @@ "task": "app.tasks.digest.send_daily_digest", "schedule": crontab(hour=12, minute=30), # 6:00 PM IST = 12:30 PM UTC }, + "reconcile-usage-periods": { + "task": "reconcile_usage_periods", + "schedule": crontab(hour=2, minute=0), # 2:00 AM UTC daily + }, } diff --git a/backend/app/main.py b/backend/app/main.py index 8b458f5..ee3167e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -8,7 +8,7 @@ from app.middleware.setup import SecurityHeadersMiddleware, RequestLoggingMiddleware from app.middleware.logging import setup_logging from app.middleware.cors import WildcardCORSMiddleware -from app.routers import auth, workspace, jobs, resumes, ai, billing, extension, webhooks, files, health +from app.routers import auth, workspace, jobs, resumes, ai, billing, extension, webhooks, files, health, admin setup_logging() logger = structlog.get_logger() @@ -51,6 +51,7 @@ async def lifespan(app: FastAPI): app.include_router(extension.router) app.include_router(webhooks.router) app.include_router(files.router) +app.include_router(admin.router) @app.get("/extension-version.json") diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index c6458ef..f3d0eb2 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -12,6 +12,7 @@ from app.models.builder_resume import BuilderResume, BuilderResumeVersion from app.models.download_log import DownloadLog from app.models.prompt_log import PromptLog +from app.models.usage import UserUsage __all__ = [ "User", @@ -37,4 +38,5 @@ "BuilderResumeVersion", "DownloadLog", "PromptLog", + "UserUsage", ] diff --git a/backend/app/models/subscription.py b/backend/app/models/subscription.py index 7905200..2dc4709 100644 --- a/backend/app/models/subscription.py +++ b/backend/app/models/subscription.py @@ -26,6 +26,10 @@ class Subscription(Base): suspended: Mapped[bool] = mapped_column(Boolean, default=False) lifetime_deal: Mapped[bool] = mapped_column(Boolean, default=False) provider: Mapped[str] = mapped_column(String, default="razorpay") + last_payment_status: Mapped[str | None] = mapped_column(String, nullable=True) + last_payment_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + paused_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + cancelled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) @@ -39,6 +43,7 @@ class PaymentEvent(Base): provider: Mapped[str] = mapped_column(String, default="razorpay") event_id: Mapped[str] = mapped_column(String, nullable=False, unique=True) event_type: Mapped[str] = mapped_column(String, nullable=False) - workspace_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False) + workspace_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=True) + user_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True) payload: Mapped[dict] = mapped_column(JSONB, default={}) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) diff --git a/backend/app/models/usage.py b/backend/app/models/usage.py new file mode 100644 index 0000000..b681dae --- /dev/null +++ b/backend/app/models/usage.py @@ -0,0 +1,20 @@ +import uuid +from datetime import date, datetime +from sqlalchemy import Date, Integer, DateTime, ForeignKey, UniqueConstraint, func +from sqlalchemy.dialects.postgresql import UUID +from sqlalchemy.orm import Mapped, mapped_column +from app.database import Base + + +class UserUsage(Base): + __tablename__ = "user_usage" + __table_args__ = (UniqueConstraint("user_id", "period_start", name="uq_user_usage_user_period"),) + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + user_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) + period_start: Mapped[date] = mapped_column(Date, nullable=False) + period_end: Mapped[date] = mapped_column(Date, nullable=False) + job_tracks_used: Mapped[int] = mapped_column(Integer, default=0) + cover_letters_used: Mapped[int] = mapped_column(Integer, default=0) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/backend/app/routers/admin.py b/backend/app/routers/admin.py new file mode 100644 index 0000000..474dc07 --- /dev/null +++ b/backend/app/routers/admin.py @@ -0,0 +1,34 @@ +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.dependencies.auth import require_admin +from app.schemas.admin import ( + AdminOverviewOut, + AdminSubscriptionOut, + AdminUsageRowOut, + RevenuePointOut, +) +from app.services import admin as admin_service + +router = APIRouter(prefix="/api/v1/admin", tags=["admin"]) + + +@router.get("/stats/overview", response_model=AdminOverviewOut, dependencies=[Depends(require_admin)]) +async def stats_overview(db: AsyncSession = Depends(get_db)): + return await admin_service.get_overview(db) + + +@router.get("/stats/revenue", response_model=list[RevenuePointOut], dependencies=[Depends(require_admin)]) +async def stats_revenue(days: int = 30, db: AsyncSession = Depends(get_db)): + return await admin_service.get_revenue_series(db, days=days) + + +@router.get("/subscriptions", response_model=list[AdminSubscriptionOut], dependencies=[Depends(require_admin)]) +async def list_subscriptions(limit: int = 50, offset: int = 0, db: AsyncSession = Depends(get_db)): + return await admin_service.list_subscriptions(db, limit=limit, offset=offset) + + +@router.get("/usage", response_model=list[AdminUsageRowOut], dependencies=[Depends(require_admin)]) +async def list_usage(limit: int = 50, db: AsyncSession = Depends(get_db)): + return await admin_service.list_usage(db, limit=limit) diff --git a/backend/app/routers/ai.py b/backend/app/routers/ai.py index 3a28c4c..1edd7de 100644 --- a/backend/app/routers/ai.py +++ b/backend/app/routers/ai.py @@ -7,8 +7,10 @@ from app.database import get_db from app.dependencies.auth import get_current_user from app.schemas.ai import DeepSeekRequest, DeepSeekResult +from app.schemas.usage import PlanLimitErrorOut from app.services import ai as ai_service from app.services import workspace as workspace_service +from app.services import usage as usage_service from app.models.job import Job router = APIRouter(prefix="/api/v1/ai", tags=["ai"]) @@ -28,10 +30,21 @@ async def ai_generate( db: AsyncSession = Depends(get_db), ): ws = await get_workspace_info(user, db) + user_id = uuid.UUID(user["user_id"]) - allowed = await ai_service.check_entitlement(db, str(ws.id), data.job_id) - if not allowed: - raise HTTPException(status_code=status.HTTP_402_PAYMENT_REQUIRED, detail="Free trial limit reached") + is_cover_letter = data.prompt_name == "generate_cover_letter" + if is_cover_letter: + check = await usage_service.can_generate_cover_letter(db, user_id) + if not check.allowed: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=PlanLimitErrorOut( + message=f"You've reached your {check.plan_name} plan's cover letter limit. Upgrade to generate more.", + currentPlan=check.plan_name, + limit=check.limit, + used=check.used, + ).model_dump(), + ) try: result = await ai_service.call_deepseek( @@ -84,6 +97,9 @@ async def ai_generate( except (json.JSONDecodeError, ValueError): pass + if is_cover_letter: + await usage_service.increment_cover_letter_usage(db, user_id) + return result @@ -93,6 +109,11 @@ async def check_entitlement( user: dict = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): - ws = await get_workspace_info(user, db) - allowed = await ai_service.check_entitlement(db, str(ws.id), job_id) - return {"allowed": allowed} + user_id = uuid.UUID(user["user_id"]) + job_tracks = await usage_service.can_create_job_track(db, user_id) + cover_letters = await usage_service.can_generate_cover_letter(db, user_id) + return { + "allowed": job_tracks.allowed and cover_letters.allowed, + "job_track": {"allowed": job_tracks.allowed, "used": job_tracks.used, "limit": job_tracks.limit}, + "cover_letter": {"allowed": cover_letters.allowed, "used": cover_letters.used, "limit": cover_letters.limit}, + } diff --git a/backend/app/routers/billing.py b/backend/app/routers/billing.py index 747f533..26207d2 100644 --- a/backend/app/routers/billing.py +++ b/backend/app/routers/billing.py @@ -1,8 +1,7 @@ import uuid import base64 -import hashlib import json -from fastapi import APIRouter, Depends, HTTPException, status, Request +from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db @@ -10,27 +9,18 @@ from app.schemas.billing import ( BillingStatusOut, CreateSubscriptionRequest, + CancelSubscriptionRequest, SubscriptionOut, + BillingHistoryItemOut, PricingOut, ) -from app.schemas.extension import ( - ExtensionTokenCreate, - ExtensionTokenOut, - ExtensionTokenCreatedOut, - ExtensionJobCreate, -) -from app.services import workspace as workspace_service -from app.services import jobs as jobs_service -from app.models.job import JobStatus +from app.schemas.usage import UsageOut +from app.services import subscription as subscription_service +from app.services import usage as usage_service from app.config import settings -from app.models.extension_token import ExtensionToken from app.models.subscription import Subscription from app.models.plan import Plan, GeoPricing -from app.models.workspace import Workspace -from app.models.job import JobArtifact, ArtifactKind -from app.models.board import Board -from sqlalchemy import select, delete -from datetime import datetime, timezone +from sqlalchemy import select import httpx router = APIRouter(prefix="/api/v1/billing", tags=["billing"]) @@ -41,54 +31,33 @@ async def billing_status( user: dict = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): - ws = await workspace_service.get_workspace_for_user(db, uuid.UUID(user["user_id"])) - if not ws: - return BillingStatusOut( - plan="free", - currency="USD", - trial_used=0, - trial_limit=2, - has_pro=False, - ) - - from app.models.workspace import Workspace as WsModel - from app.models.subscription import Subscription as SubModel + user_id = uuid.UUID(user["user_id"]) + sub = await subscription_service.get_subscription_for_user(db, user_id) + plan = await usage_service.get_active_plan(db, user_id, sub=sub) + job_tracks = await usage_service.can_create_job_track(db, user_id) - ws_result = await db.execute( - select(WsModel).where(WsModel.owner_user_id == uuid.UUID(user["user_id"])).limit(1) - ) - ws = ws_result.scalar_one_or_none() - - is_pro = False - if ws: - sub_result = await db.execute(select(SubModel).where(SubModel.user_id == uuid.UUID(user["user_id"]))) - sub = sub_result.scalar_one_or_none() - if sub and sub.subscription_status == "active" and not sub.suspended: - from datetime import timezone as tz - if sub.current_period_end is None or sub.current_period_end > datetime.now(tz.utc): - is_pro = True - - art_result = await db.execute( - select(JobArtifact) - .where(JobArtifact.workspace_id == ws.id) - .where(JobArtifact.kind.in_([ArtifactKind.TAILORED_RESUME, ArtifactKind.COVER_LETTER])) - ) - arts = list(art_result.scalars().all()) - distinct_jobs = set(str(a.job_id) for a in arts if a.job_id) - - sub_result = await db.execute(select(SubModel).where(SubModel.user_id == uuid.UUID(user["user_id"]))) - sub = sub_result.scalar_one_or_none() + is_pro = plan.id != "free" return BillingStatusOut( - plan=ws.plan if ws else "free", - currency=ws.currency if ws else "USD", - trial_used=len(distinct_jobs), - trial_limit=int(ws.trial_apps_limit) if ws else 2, + plan=plan.id, + currency="USD", + trial_used=job_tracks.used, + trial_limit=job_tracks.limit if job_tracks.limit is not None else -1, has_pro=is_pro, current_period_end=sub.current_period_end if sub else None, + cancel_at_period_end=bool(sub.cancel_at_period_end) if sub else False, ) +@router.get("/usage", response_model=UsageOut) +async def get_usage( + user: dict = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + snapshot = await usage_service.get_usage_snapshot(db, uuid.UUID(user["user_id"])) + return UsageOut(**snapshot) + + @router.post("/subscription/create") async def create_subscription( data: CreateSubscriptionRequest, @@ -101,28 +70,16 @@ async def create_subscription( raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Razorpay not configured") country_code = data.country_code.upper() + user_id = uuid.UUID(user["user_id"]) - pricing_result = await db.execute( - select(GeoPricing) - .where(GeoPricing.plan_id == data.plan_id) - .where(GeoPricing.country_code.in_([country_code, "DEFAULT"])) - ) - rows = list(pricing_result.scalars().all()) - chosen = next((r for r in rows if r.country_code == country_code), None) - if not chosen: - chosen = next((r for r in rows if r.country_code == "DEFAULT"), None) - - razorpay_plan_id = ( - chosen.razorpay_plan_id_annual if data.billing_cycle == "annual" else chosen.razorpay_plan_id_monthly + # Server-side resolution only — never trust a client-supplied plan/price. + razorpay_plan_id = await subscription_service.resolve_razorpay_plan_id( + db, data.plan_id, data.billing_cycle, country_code ) if not razorpay_plan_id: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Plan not available in your region") - existing_result = await db.execute( - select(Subscription).where(Subscription.user_id == uuid.UUID(user["user_id"])) - ) - existing = existing_result.scalar_one_or_none() - + existing = await subscription_service.get_subscription_for_user(db, user_id) if existing and existing.razorpay_subscription_id and existing.subscription_status == "active": return { "subscription_id": existing.razorpay_subscription_id, @@ -136,7 +93,7 @@ async def create_subscription( "total_count": 10 if data.billing_cycle == "annual" else 120, "customer_notify": 1, "notes": { - "user_id": user["user_id"], + "user_id": str(user_id), "plan_id": data.plan_id, "billing_cycle": data.billing_cycle, "country_code": country_code, @@ -159,24 +116,9 @@ async def create_subscription( rp_data = json.loads(text) rp_sub_id = rp_data.get("id") - if existing: - existing.plan = data.plan_id - existing.plan_id = data.plan_id - existing.razorpay_subscription_id = rp_sub_id - existing.subscription_status = rp_data.get("status") - existing.billing_cycle = data.billing_cycle - else: - sub = Subscription( - user_id=uuid.UUID(user["user_id"]), - plan=data.plan_id, - plan_id=data.plan_id, - razorpay_subscription_id=rp_sub_id, - subscription_status=rp_data.get("status"), - billing_cycle=data.billing_cycle, - ) - db.add(sub) - - await db.flush() + await subscription_service.create_or_update_subscription_from_razorpay( + db, user_id, data.plan_id, data.billing_cycle, rp_sub_id, rp_data.get("status") + ) return { "subscription_id": rp_sub_id, @@ -187,6 +129,35 @@ async def create_subscription( } +@router.post("/subscription/cancel", response_model=SubscriptionOut) +async def cancel_subscription( + data: CancelSubscriptionRequest, + user: dict = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + try: + sub = await subscription_service.cancel_subscription(db, uuid.UUID(user["user_id"]), data.at_period_end) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(e)) + return sub + + +@router.post("/subscription/reactivate", response_model=SubscriptionOut) +async def reactivate_subscription( + user: dict = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + try: + sub = await subscription_service.reactivate_subscription(db, uuid.UUID(user["user_id"])) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(e)) + return sub + + @router.get("/subscription", response_model=SubscriptionOut | None) async def get_subscription( user: dict = Depends(get_current_user), @@ -196,6 +167,35 @@ async def get_subscription( return result.scalar_one_or_none() +_EVENT_SUMMARIES = { + "subscription.activated": "Subscription activated", + "subscription.charged": "Payment charged — subscription renewed", + "subscription.completed": "Subscription completed", + "subscription.cancelled": "Subscription cancelled", + "subscription.paused": "Subscription paused", + "subscription.resumed": "Subscription resumed", + "payment.authorized": "Payment authorized", + "payment.failed": "Payment failed", +} + + +@router.get("/subscription/history", response_model=list[BillingHistoryItemOut]) +async def get_billing_history( + user: dict = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + events = await subscription_service.list_billing_history(db, uuid.UUID(user["user_id"])) + return [ + BillingHistoryItemOut( + id=e.id, + event_type=e.event_type, + created_at=e.created_at, + summary=_EVENT_SUMMARIES.get(e.event_type, e.event_type), + ) + for e in events + ] + + @router.get("/pricing", response_model=list[PricingOut]) async def get_pricing( country_code: str = "DEFAULT", diff --git a/backend/app/routers/jobs.py b/backend/app/routers/jobs.py index 8220bd3..5fab88f 100644 --- a/backend/app/routers/jobs.py +++ b/backend/app/routers/jobs.py @@ -18,11 +18,27 @@ from app.models.job import JobStatus, ArtifactKind from app.services import jobs as jobs_service from app.services import workspace as workspace_service +from app.services import usage as usage_service +from app.schemas.usage import PlanLimitErrorOut from pydantic import BaseModel router = APIRouter(prefix="/api/v1/jobs", tags=["jobs"]) +async def _raise_if_job_track_blocked(db: AsyncSession, user_id: uuid.UUID) -> None: + check = await usage_service.can_create_job_track(db, user_id) + if not check.allowed: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=PlanLimitErrorOut( + message=f"You've reached your {check.plan_name} plan's job track limit. Upgrade to add more.", + currentPlan=check.plan_name, + limit=check.limit, + used=check.used, + ).model_dump(), + ) + + async def get_workspace_id(user: dict, db: AsyncSession) -> uuid.UUID: ws = await workspace_service.get_workspace_for_user(db, uuid.UUID(user["user_id"])) if not ws: @@ -57,8 +73,12 @@ async def create_job( user: dict = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): + user_id = uuid.UUID(user["user_id"]) ws_id = await get_workspace_id(user, db) - return await jobs_service.create_job( + + await _raise_if_job_track_blocked(db, user_id) + + job = await jobs_service.create_job( db, workspace_id=ws_id, board_id=data.board_id, @@ -72,6 +92,8 @@ async def create_job( date_applied=str(data.date_applied) if data.date_applied else None, resume_score=data.resume_score, ) + await usage_service.increment_job_track_usage(db, user_id) + return job @router.put("/{job_id}") diff --git a/backend/app/routers/webhooks.py b/backend/app/routers/webhooks.py index 06a25c6..a2bf665 100644 --- a/backend/app/routers/webhooks.py +++ b/backend/app/routers/webhooks.py @@ -1,14 +1,13 @@ import json import hmac import hashlib -import base64 from fastapi import APIRouter, Depends, HTTPException, status, Request from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.config import settings -from app.models.subscription import Subscription, PaymentEvent -from app.models.workspace import Workspace +from app.models.subscription import PaymentEvent +from app.services.webhook import WEBHOOK_HANDLERS, record_payment_event from sqlalchemy import select router = APIRouter(prefix="/api/v1/webhooks", tags=["webhooks"]) @@ -37,6 +36,7 @@ async def razorpay_webhook(request: Request, db: AsyncSession = Depends(get_db)) event_id = payload.get("event_id", "") event_type = payload.get("event", "") + # Idempotency: skip if we've already processed this exact event. if event_id: existing = await db.execute( select(PaymentEvent).where(PaymentEvent.event_id == event_id) @@ -44,45 +44,9 @@ async def razorpay_webhook(request: Request, db: AsyncSession = Depends(get_db)) if existing.scalar_one_or_none(): return {"ok": True, "skipped": "duplicate"} - sub_entity = (payload.get("payload", {}) or {}).get("subscription", {}) or {} - entity = (sub_entity.get("entity", {}) or {}) + handler = WEBHOOK_HANDLERS.get(event_type) + user_id = await handler(db, payload) if handler else None - rp_sub_id = entity.get("id") or sub_entity.get("id") - status_val = entity.get("status") or sub_entity.get("status") - notes = entity.get("notes", {}) or {} + await record_payment_event(db, event_id, event_type, payload, user_id) - user_id = notes.get("user_id") - - if user_id and rp_sub_id: - import uuid - sub_result = await db.execute( - select(Subscription).where(Subscription.razorpay_subscription_id == rp_sub_id) - ) - sub = sub_result.scalar_one_or_none() - plan_from_notes = notes.get("plan_id", "pro") - if sub: - sub.subscription_status = status_val - if sub.plan == "free": - sub.plan = plan_from_notes - sub.plan_id = plan_from_notes - else: - sub = Subscription( - user_id=uuid.UUID(user_id), - plan=plan_from_notes, - plan_id=plan_from_notes, - razorpay_subscription_id=rp_sub_id, - subscription_status=status_val, - ) - db.add(sub) - - event = PaymentEvent( - provider="razorpay", - event_id=event_id, - event_type=event_type, - workspace_id=uuid.UUID("00000000-0000-0000-0000-000000000000"), - payload=payload, - ) - db.add(event) - await db.flush() - - return {"ok": True} + return {"ok": True, "handled": handler is not None} diff --git a/backend/app/schemas/admin.py b/backend/app/schemas/admin.py new file mode 100644 index 0000000..e6821fb --- /dev/null +++ b/backend/app/schemas/admin.py @@ -0,0 +1,51 @@ +from pydantic import BaseModel +from typing import Optional +from uuid import UUID +from datetime import datetime + + +class PlanDistributionOut(BaseModel): + plan_id: str + plan_name: str + count: int + + +class AdminOverviewOut(BaseModel): + total_users: int + active_subscribers: int + cancelled_subscribers: int + suspended_subscribers: int + free_users: int + plan_distribution: list[PlanDistributionOut] + mrr_usd: float + + +class RevenuePointOut(BaseModel): + date: str + amount_usd: float + event_count: int + + +class AdminSubscriptionOut(BaseModel): + id: UUID + user_id: UUID + email: str + full_name: Optional[str] = None + plan: str + subscription_status: Optional[str] = None + billing_cycle: Optional[str] = None + current_period_end: Optional[datetime] = None + cancel_at_period_end: bool + suspended: bool + created_at: datetime + + model_config = {"from_attributes": True} + + +class AdminUsageRowOut(BaseModel): + user_id: UUID + email: str + plan: str + job_tracks_used: int + cover_letters_used: int + period_start: str diff --git a/backend/app/schemas/billing.py b/backend/app/schemas/billing.py index 5e6d898..9d293d2 100644 --- a/backend/app/schemas/billing.py +++ b/backend/app/schemas/billing.py @@ -1,5 +1,5 @@ from pydantic import BaseModel, Field -from typing import Optional +from typing import Optional, Any from uuid import UUID from datetime import datetime @@ -11,6 +11,7 @@ class BillingStatusOut(BaseModel): trial_limit: int has_pro: bool current_period_end: Optional[datetime] = None + cancel_at_period_end: bool = False class CreateSubscriptionRequest(BaseModel): @@ -20,6 +21,10 @@ class CreateSubscriptionRequest(BaseModel): trial: bool = False +class CancelSubscriptionRequest(BaseModel): + at_period_end: bool = True + + class SubscriptionOut(BaseModel): id: UUID user_id: UUID @@ -27,10 +32,25 @@ class SubscriptionOut(BaseModel): subscription_status: Optional[str] = None razorpay_subscription_id: Optional[str] = None billing_cycle: Optional[str] = None + current_period_start: Optional[datetime] = None current_period_end: Optional[datetime] = None + cancel_at_period_end: bool = False trial_ends_at: Optional[datetime] = None suspended: bool + last_payment_status: Optional[str] = None + last_payment_at: Optional[datetime] = None + paused_at: Optional[datetime] = None + cancelled_at: Optional[datetime] = None + created_at: datetime + + model_config = {"from_attributes": True} + + +class BillingHistoryItemOut(BaseModel): + id: UUID + event_type: str created_at: datetime + summary: str model_config = {"from_attributes": True} diff --git a/backend/app/schemas/usage.py b/backend/app/schemas/usage.py new file mode 100644 index 0000000..c7eebb8 --- /dev/null +++ b/backend/app/schemas/usage.py @@ -0,0 +1,27 @@ +from pydantic import BaseModel +from datetime import date +from typing import Literal, Optional + + +class UsageMetricOut(BaseModel): + used: int + limit: Optional[int] = None + remaining: Optional[int] = None + unlimited: bool = False + + +class UsageOut(BaseModel): + plan: str + period_start: date + period_end: date + job_tracks: UsageMetricOut + cover_letters: UsageMetricOut + + +class PlanLimitErrorOut(BaseModel): + success: bool = False + reason: Literal["PLAN_LIMIT_REACHED"] = "PLAN_LIMIT_REACHED" + message: str + currentPlan: str + limit: Optional[int] = None + used: int diff --git a/backend/app/services/admin.py b/backend/app/services/admin.py new file mode 100644 index 0000000..79f8e3c --- /dev/null +++ b/backend/app/services/admin.py @@ -0,0 +1,142 @@ +import uuid +from datetime import date, datetime, timedelta, timezone + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.plan import Plan +from app.models.subscription import PaymentEvent, Subscription +from app.models.usage import UserUsage +from app.models.user import User + + +async def get_overview(db: AsyncSession) -> dict: + total_users = (await db.execute(select(func.count(User.id)))).scalar_one() + + active = ( + await db.execute( + select(func.count(Subscription.id)).where( + Subscription.subscription_status == "active", Subscription.suspended.is_(False) + ) + ) + ).scalar_one() + cancelled = ( + await db.execute(select(func.count(Subscription.id)).where(Subscription.subscription_status == "cancelled")) + ).scalar_one() + suspended = (await db.execute(select(func.count(Subscription.id)).where(Subscription.suspended.is_(True)))).scalar_one() + + dist_result = await db.execute( + select(Subscription.plan_id, Plan.name, func.count(Subscription.id)) + .join(Plan, Plan.id == Subscription.plan_id, isouter=True) + .group_by(Subscription.plan_id, Plan.name) + ) + plan_distribution = [ + {"plan_id": plan_id or "free", "plan_name": plan_name or "Free", "count": count} + for plan_id, plan_name, count in dist_result.all() + ] + free_users = next((row["count"] for row in plan_distribution if row["plan_id"] == "free"), 0) + + # MRR: active, non-suspended paid subscriptions normalized to a monthly figure. + mrr_result = await db.execute( + select(Subscription.billing_cycle, Plan.monthly_price_usd, Plan.annual_price_usd) + .join(Plan, Plan.id == Subscription.plan_id) + .where( + Subscription.subscription_status == "active", + Subscription.suspended.is_(False), + Subscription.plan_id != "free", + ) + ) + mrr = 0.0 + for billing_cycle, monthly_price, annual_price in mrr_result.all(): + if billing_cycle == "annual" and annual_price: + mrr += float(annual_price) / 12 + elif monthly_price: + mrr += float(monthly_price) + + return { + "total_users": total_users, + "active_subscribers": active, + "cancelled_subscribers": cancelled, + "suspended_subscribers": suspended, + "free_users": free_users, + "plan_distribution": plan_distribution, + "mrr_usd": round(mrr, 2), + } + + +async def get_revenue_series(db: AsyncSession, days: int = 30) -> list[dict]: + since = datetime.now(timezone.utc) - timedelta(days=days) + result = await db.execute( + select(PaymentEvent) + .where(PaymentEvent.event_type.in_(["subscription.charged", "payment.authorized"])) + .where(PaymentEvent.created_at >= since) + .order_by(PaymentEvent.created_at) + ) + events = list(result.scalars().all()) + + buckets: dict[str, dict] = {} + for e in events: + day = e.created_at.date().isoformat() + entity = ((e.payload.get("payload", {}) or {}).get("payment", {}) or {}).get("entity", {}) or {} + amount = (entity.get("amount") or 0) / 100 # Razorpay amounts are in the smallest currency unit + bucket = buckets.setdefault(day, {"amount_usd": 0.0, "event_count": 0}) + bucket["amount_usd"] += amount + bucket["event_count"] += 1 + + return [ + {"date": day, "amount_usd": round(v["amount_usd"], 2), "event_count": v["event_count"]} + for day, v in sorted(buckets.items()) + ] + + +async def list_subscriptions(db: AsyncSession, limit: int = 50, offset: int = 0) -> list[dict]: + result = await db.execute( + select(Subscription, User) + .join(User, User.id == Subscription.user_id) + .order_by(Subscription.created_at.desc()) + .limit(limit) + .offset(offset) + ) + rows = [] + for sub, user in result.all(): + rows.append( + { + "id": sub.id, + "user_id": sub.user_id, + "email": user.email, + "full_name": user.full_name, + "plan": sub.plan_id or sub.plan or "free", + "subscription_status": sub.subscription_status, + "billing_cycle": sub.billing_cycle, + "current_period_end": sub.current_period_end, + "cancel_at_period_end": sub.cancel_at_period_end, + "suspended": sub.suspended, + "created_at": sub.created_at, + } + ) + return rows + + +async def list_usage(db: AsyncSession, limit: int = 50) -> list[dict]: + today = date.today() + result = await db.execute( + select(UserUsage, User, Subscription) + .join(User, User.id == UserUsage.user_id) + .join(Subscription, Subscription.user_id == UserUsage.user_id, isouter=True) + .where(UserUsage.period_start <= today, UserUsage.period_end > today) + .order_by((UserUsage.job_tracks_used + UserUsage.cover_letters_used).desc()) + .limit(limit) + ) + rows = [] + for usage, user, sub in result.all(): + rows.append( + { + "user_id": usage.user_id, + "email": user.email, + "plan": (sub.plan_id if sub else None) or "free", + "job_tracks_used": usage.job_tracks_used, + "cover_letters_used": usage.cover_letters_used, + "period_start": usage.period_start.isoformat(), + } + ) + return rows diff --git a/backend/app/services/ai.py b/backend/app/services/ai.py index d92e705..d903399 100644 --- a/backend/app/services/ai.py +++ b/backend/app/services/ai.py @@ -186,41 +186,3 @@ async def log_ai_cost( await db.flush() -async def check_entitlement(db: AsyncSession, workspace_id: str, job_id: str | None = None) -> bool: - from app.models.workspace import Workspace - from app.models.job import JobArtifact, ArtifactKind - from sqlalchemy import select - import uuid as _uuid - - ws_result = await db.execute( - select(Workspace).where(Workspace.id == _uuid.UUID(workspace_id)) - ) - ws = ws_result.scalar_one_or_none() - if not ws: - return False - - if ws.plan == "pro": - return True - - from app.models.subscription import Subscription - - sub_result = await db.execute( - select(Subscription).where(Subscription.user_id == ws.owner_user_id) - ) - sub = sub_result.scalar_one_or_none() - if sub and sub.subscription_status == "active" and not sub.suspended: - return True - - art_result = await db.execute( - select(JobArtifact) - .where(JobArtifact.workspace_id == _uuid.UUID(workspace_id)) - .where(JobArtifact.kind.in_([ArtifactKind.TAILORED_RESUME, ArtifactKind.COVER_LETTER])) - ) - arts = list(art_result.scalars().all()) - - distinct_jobs = set(str(a.job_id) for a in arts if a.job_id) - if job_id and job_id in distinct_jobs: - return True - - limit = ws.trial_apps_limit - return len(distinct_jobs) < limit diff --git a/backend/app/services/subscription.py b/backend/app/services/subscription.py new file mode 100644 index 0000000..13c05d1 --- /dev/null +++ b/backend/app/services/subscription.py @@ -0,0 +1,140 @@ +import base64 +import uuid +from datetime import datetime, timezone + +import httpx +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.models.plan import GeoPricing +from app.models.subscription import PaymentEvent, Subscription + +RAZORPAY_API_BASE = "https://api.razorpay.com/v1" + + +def _auth_header() -> str: + key_id = settings.RAZORPAY_KEY_ID.strip() + key_secret = settings.RAZORPAY_KEY_SECRET.strip() + return base64.b64encode(f"{key_id}:{key_secret}".encode()).decode() + + +def _razorpay_configured() -> bool: + return bool(settings.RAZORPAY_KEY_ID.strip() and settings.RAZORPAY_KEY_SECRET.strip()) + + +async def get_subscription_for_user(db: AsyncSession, user_id: uuid.UUID) -> Subscription | None: + result = await db.execute(select(Subscription).where(Subscription.user_id == user_id)) + return result.scalar_one_or_none() + + +async def resolve_razorpay_plan_id( + db: AsyncSession, plan_id: str, billing_cycle: str, country_code: str +) -> str | None: + """Server-side lookup of the real Razorpay plan for a (plan, cycle, country) — + never trust a client-supplied Razorpay plan id or price.""" + result = await db.execute( + select(GeoPricing) + .where(GeoPricing.plan_id == plan_id) + .where(GeoPricing.country_code.in_([country_code, "DEFAULT"])) + ) + rows = list(result.scalars().all()) + chosen = next((r for r in rows if r.country_code == country_code), None) + if not chosen: + chosen = next((r for r in rows if r.country_code == "DEFAULT"), None) + if not chosen: + return None + return chosen.razorpay_plan_id_annual if billing_cycle == "annual" else chosen.razorpay_plan_id_monthly + + +async def create_or_update_subscription_from_razorpay( + db: AsyncSession, + user_id: uuid.UUID, + plan_id: str, + billing_cycle: str, + rp_subscription_id: str, + rp_status: str | None, +) -> Subscription: + existing = await get_subscription_for_user(db, user_id) + if existing: + existing.plan = plan_id + existing.plan_id = plan_id + existing.razorpay_subscription_id = rp_subscription_id + existing.subscription_status = rp_status + existing.billing_cycle = billing_cycle + existing.cancel_at_period_end = False + existing.cancelled_at = None + await db.flush() + return existing + + sub = Subscription( + user_id=user_id, + plan=plan_id, + plan_id=plan_id, + razorpay_subscription_id=rp_subscription_id, + subscription_status=rp_status, + billing_cycle=billing_cycle, + ) + db.add(sub) + await db.flush() + return sub + + +async def cancel_subscription(db: AsyncSession, user_id: uuid.UUID, at_period_end: bool = True) -> Subscription: + sub = await get_subscription_for_user(db, user_id) + if not sub or not sub.razorpay_subscription_id or sub.plan_id == "free": + raise ValueError("No active paid subscription to cancel") + + if _razorpay_configured(): + async with httpx.AsyncClient(timeout=30) as client: + res = await client.post( + f"{RAZORPAY_API_BASE}/subscriptions/{sub.razorpay_subscription_id}/cancel", + headers={"Content-Type": "application/json", "Authorization": f"Basic {_auth_header()}"}, + json={"cancel_at_cycle_end": 1 if at_period_end else 0}, + ) + if res.status_code not in (200, 201): + raise RuntimeError(f"Razorpay cancel failed: {res.text[:500]}") + + if at_period_end: + sub.cancel_at_period_end = True + else: + sub.subscription_status = "cancelled" + sub.cancelled_at = datetime.now(timezone.utc) + sub.cancel_at_period_end = False + + await db.flush() + return sub + + +async def reactivate_subscription(db: AsyncSession, user_id: uuid.UUID) -> Subscription: + sub = await get_subscription_for_user(db, user_id) + if not sub: + raise ValueError("No subscription found") + if not sub.cancel_at_period_end: + raise ValueError("Subscription is not pending cancellation") + if sub.current_period_end and sub.current_period_end <= datetime.now(timezone.utc): + raise ValueError("Billing period has already ended — start a new subscription instead") + + if _razorpay_configured() and sub.razorpay_subscription_id: + async with httpx.AsyncClient(timeout=30) as client: + res = await client.patch( + f"{RAZORPAY_API_BASE}/subscriptions/{sub.razorpay_subscription_id}", + headers={"Content-Type": "application/json", "Authorization": f"Basic {_auth_header()}"}, + json={"cancel_at_cycle_end": 0}, + ) + if res.status_code not in (200, 201): + raise RuntimeError(f"Razorpay reactivate failed: {res.text[:500]}") + + sub.cancel_at_period_end = False + await db.flush() + return sub + + +async def list_billing_history(db: AsyncSession, user_id: uuid.UUID, limit: int = 50) -> list[PaymentEvent]: + result = await db.execute( + select(PaymentEvent) + .where(PaymentEvent.user_id == user_id) + .order_by(PaymentEvent.created_at.desc()) + .limit(limit) + ) + return list(result.scalars().all()) diff --git a/backend/app/services/usage.py b/backend/app/services/usage.py new file mode 100644 index 0000000..abd0a1b --- /dev/null +++ b/backend/app/services/usage.py @@ -0,0 +1,150 @@ +import uuid +from dataclasses import dataclass +from datetime import date, datetime, timezone + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.plan import Plan +from app.models.subscription import Subscription +from app.models.usage import UserUsage + + +@dataclass +class UsageCheckResult: + allowed: bool + used: int + limit: int | None + remaining: int | None + plan_name: str + plan_id: str + + +def _is_entitled(sub: Subscription) -> bool: + """Whether a subscription currently grants its plan's access (paid tiers only). + + Cancellation doesn't revoke access immediately — Razorpay-style billing keeps + access until `current_period_end` lapses. Pausing/suspension does revoke it now. + """ + if sub.suspended: + return False + if sub.subscription_status == "paused": + return False + if sub.subscription_status in ("active", "cancelled", "authenticated"): + if sub.current_period_end is None: + return True + return sub.current_period_end > datetime.now(timezone.utc) + return False + + +async def get_subscription_for_user(db: AsyncSession, user_id: uuid.UUID) -> Subscription | None: + result = await db.execute(select(Subscription).where(Subscription.user_id == user_id)) + return result.scalar_one_or_none() + + +async def get_active_plan(db: AsyncSession, user_id: uuid.UUID, sub: Subscription | None = None) -> Plan: + if sub is None: + sub = await get_subscription_for_user(db, user_id) + + plan_id = "free" + if sub and sub.plan_id and sub.plan_id != "free" and _is_entitled(sub): + plan_id = sub.plan_id + + result = await db.execute(select(Plan).where(Plan.id == plan_id)) + plan = result.scalar_one_or_none() + if not plan: + result = await db.execute(select(Plan).where(Plan.id == "free")) + plan = result.scalar_one_or_none() + if not plan: + raise RuntimeError("No 'free' plan row found — check plans table seed data") + return plan + + +def get_current_period(sub: Subscription | None) -> tuple[date, date]: + """Paid, entitled subscriptions use their own Razorpay billing cycle; everyone + else (free plan, lapsed/inactive subscriptions) uses the calendar month.""" + if sub and sub.current_period_start and sub.current_period_end and _is_entitled(sub): + return sub.current_period_start.date(), sub.current_period_end.date() + + now = datetime.now(timezone.utc) + start = date(now.year, now.month, 1) + end = date(now.year + 1, 1, 1) if now.month == 12 else date(now.year, now.month + 1, 1) + return start, end + + +async def get_or_create_usage( + db: AsyncSession, user_id: uuid.UUID, period_start: date, period_end: date +) -> UserUsage: + result = await db.execute( + select(UserUsage).where(UserUsage.user_id == user_id, UserUsage.period_start == period_start) + ) + usage = result.scalar_one_or_none() + if usage: + return usage + + usage = UserUsage(user_id=user_id, period_start=period_start, period_end=period_end) + db.add(usage) + await db.flush() + return usage + + +async def _check(db: AsyncSession, user_id: uuid.UUID, metric: str) -> UsageCheckResult: + sub = await get_subscription_for_user(db, user_id) + plan = await get_active_plan(db, user_id, sub=sub) + limit = plan.job_track_limit if metric == "job_tracks" else plan.cover_letter_limit + + if limit is None: + return UsageCheckResult(allowed=True, used=0, limit=None, remaining=None, plan_name=plan.name, plan_id=plan.id) + + period_start, period_end = get_current_period(sub) + usage = await get_or_create_usage(db, user_id, period_start, period_end) + used = usage.job_tracks_used if metric == "job_tracks" else usage.cover_letters_used + remaining = max(0, limit - used) + return UsageCheckResult( + allowed=used < limit, used=used, limit=limit, remaining=remaining, plan_name=plan.name, plan_id=plan.id + ) + + +async def can_create_job_track(db: AsyncSession, user_id: uuid.UUID) -> UsageCheckResult: + return await _check(db, user_id, "job_tracks") + + +async def can_generate_cover_letter(db: AsyncSession, user_id: uuid.UUID) -> UsageCheckResult: + return await _check(db, user_id, "cover_letters") + + +async def _increment(db: AsyncSession, user_id: uuid.UUID, metric: str) -> None: + sub = await get_subscription_for_user(db, user_id) + period_start, period_end = get_current_period(sub) + usage = await get_or_create_usage(db, user_id, period_start, period_end) + if metric == "job_tracks": + usage.job_tracks_used += 1 + else: + usage.cover_letters_used += 1 + await db.flush() + + +async def increment_job_track_usage(db: AsyncSession, user_id: uuid.UUID) -> None: + await _increment(db, user_id, "job_tracks") + + +async def increment_cover_letter_usage(db: AsyncSession, user_id: uuid.UUID) -> None: + await _increment(db, user_id, "cover_letters") + + +async def get_usage_snapshot(db: AsyncSession, user_id: uuid.UUID) -> dict: + job_tracks = await can_create_job_track(db, user_id) + cover_letters = await can_generate_cover_letter(db, user_id) + sub = await get_subscription_for_user(db, user_id) + period_start, period_end = get_current_period(sub) + + def _metric(r: UsageCheckResult) -> dict: + return {"used": r.used, "limit": r.limit, "remaining": r.remaining, "unlimited": r.limit is None} + + return { + "plan": job_tracks.plan_name, + "period_start": period_start, + "period_end": period_end, + "job_tracks": _metric(job_tracks), + "cover_letters": _metric(cover_letters), + } diff --git a/backend/app/services/webhook.py b/backend/app/services/webhook.py new file mode 100644 index 0000000..f591c26 --- /dev/null +++ b/backend/app/services/webhook.py @@ -0,0 +1,210 @@ +import uuid +from datetime import datetime, timezone + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.subscription import PaymentEvent, Subscription +from app.models.workspace import Workspace +from app.services import usage as usage_service + + +def _extract_entity(payload: dict, key: str) -> dict: + return ((payload.get("payload", {}) or {}).get(key, {}) or {}).get("entity", {}) or {} + + +def _resolve_user_id(entity: dict) -> uuid.UUID | None: + notes = entity.get("notes", {}) or {} + raw = notes.get("user_id") + if not raw: + return None + try: + return uuid.UUID(raw) + except ValueError: + return None + + +def _to_dt(epoch: int | None) -> datetime | None: + return datetime.fromtimestamp(epoch, tz=timezone.utc) if epoch else None + + +async def _find_subscription(db: AsyncSession, rp_sub_id: str | None) -> Subscription | None: + if not rp_sub_id: + return None + result = await db.execute(select(Subscription).where(Subscription.razorpay_subscription_id == rp_sub_id)) + return result.scalar_one_or_none() + + +async def _find_subscription_by_customer(db: AsyncSession, customer_id: str | None) -> Subscription | None: + if not customer_id: + return None + result = await db.execute(select(Subscription).where(Subscription.razorpay_customer_id == customer_id)) + return result.scalar_one_or_none() + + +async def _get_or_create_subscription( + db: AsyncSession, user_id: uuid.UUID | None, rp_sub_id: str | None, plan_from_notes: str +) -> Subscription | None: + sub = await _find_subscription(db, rp_sub_id) + if sub or not rp_sub_id or not user_id: + return sub + sub = Subscription(user_id=user_id, plan=plan_from_notes, plan_id=plan_from_notes, razorpay_subscription_id=rp_sub_id) + db.add(sub) + await db.flush() + return sub + + +# --- subscription.* handlers ------------------------------------------------- + +async def handle_subscription_activated(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "subscription") + user_id = _resolve_user_id(entity) + notes = entity.get("notes", {}) or {} + sub = await _get_or_create_subscription(db, user_id, entity.get("id"), notes.get("plan_id", "pro")) + if sub: + sub.subscription_status = "active" + sub.current_period_start = _to_dt(entity.get("current_start")) or sub.current_period_start + sub.current_period_end = _to_dt(entity.get("current_end")) or sub.current_period_end + if entity.get("customer_id"): + sub.razorpay_customer_id = entity["customer_id"] + sub.cancel_at_period_end = False + sub.cancelled_at = None + await db.flush() + user_id = user_id or sub.user_id + return user_id + + +async def handle_subscription_charged(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "subscription") + user_id = _resolve_user_id(entity) + notes = entity.get("notes", {}) or {} + sub = await _get_or_create_subscription(db, user_id, entity.get("id"), notes.get("plan_id", "pro")) + if sub: + sub.subscription_status = "active" + sub.last_payment_status = "authorized" + sub.last_payment_at = datetime.now(timezone.utc) + sub.current_period_start = _to_dt(entity.get("current_start")) or sub.current_period_start + sub.current_period_end = _to_dt(entity.get("current_end")) or sub.current_period_end + await db.flush() + # Pre-create the new period's usage row so dashboards don't show a transient + # empty state right at rollover. + if sub.current_period_start and sub.current_period_end: + await usage_service.get_or_create_usage( + db, sub.user_id, sub.current_period_start.date(), sub.current_period_end.date() + ) + user_id = user_id or sub.user_id + return user_id + + +async def handle_subscription_completed(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "subscription") + user_id = _resolve_user_id(entity) + sub = await _find_subscription(db, entity.get("id")) + if sub: + sub.subscription_status = "completed" + await db.flush() + user_id = user_id or sub.user_id + return user_id + + +async def handle_subscription_cancelled(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "subscription") + user_id = _resolve_user_id(entity) + sub = await _find_subscription(db, entity.get("id")) + if sub: + # Access continues until current_period_end naturally lapses — no forced downgrade here. + sub.subscription_status = "cancelled" + sub.cancelled_at = datetime.now(timezone.utc) + await db.flush() + user_id = user_id or sub.user_id + return user_id + + +async def handle_subscription_paused(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "subscription") + user_id = _resolve_user_id(entity) + sub = await _find_subscription(db, entity.get("id")) + if sub: + sub.subscription_status = "paused" + sub.paused_at = datetime.now(timezone.utc) + await db.flush() + user_id = user_id or sub.user_id + return user_id + + +async def handle_subscription_resumed(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "subscription") + user_id = _resolve_user_id(entity) + sub = await _find_subscription(db, entity.get("id")) + if sub: + sub.subscription_status = "active" + sub.paused_at = None + await db.flush() + user_id = user_id or sub.user_id + return user_id + + +# --- payment.* handlers (informational — status transitions are driven by the +# subscription.* events above, not by bare payment events) ------------------- + +async def handle_payment_authorized(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "payment") + user_id = _resolve_user_id(entity) + sub = await usage_service.get_subscription_for_user(db, user_id) if user_id else None + if not sub: + sub = await _find_subscription_by_customer(db, entity.get("customer_id")) + if sub: + sub.last_payment_status = "authorized" + sub.last_payment_at = datetime.now(timezone.utc) + await db.flush() + user_id = user_id or sub.user_id + return user_id + + +async def handle_payment_failed(db: AsyncSession, payload: dict) -> uuid.UUID | None: + entity = _extract_entity(payload, "payment") + user_id = _resolve_user_id(entity) + sub = await usage_service.get_subscription_for_user(db, user_id) if user_id else None + if not sub: + sub = await _find_subscription_by_customer(db, entity.get("customer_id")) + if sub: + sub.last_payment_status = "failed" + sub.last_payment_at = datetime.now(timezone.utc) + await db.flush() + user_id = user_id or sub.user_id + return user_id + + +WEBHOOK_HANDLERS = { + "subscription.activated": handle_subscription_activated, + "subscription.charged": handle_subscription_charged, + "subscription.completed": handle_subscription_completed, + "subscription.cancelled": handle_subscription_cancelled, + "subscription.paused": handle_subscription_paused, + "subscription.resumed": handle_subscription_resumed, + "payment.authorized": handle_payment_authorized, + "payment.failed": handle_payment_failed, +} + + +async def record_payment_event( + db: AsyncSession, event_id: str, event_type: str, payload: dict, user_id: uuid.UUID | None +) -> PaymentEvent: + workspace_id = None + if user_id: + result = await db.execute(select(Workspace).where(Workspace.owner_user_id == user_id).limit(1)) + ws = result.scalar_one_or_none() + if ws: + workspace_id = ws.id + + event = PaymentEvent( + provider="razorpay", + event_id=event_id, + event_type=event_type, + user_id=user_id, + workspace_id=workspace_id, + payload=payload, + ) + db.add(event) + await db.flush() + return event diff --git a/backend/app/tasks/usage_reset.py b/backend/app/tasks/usage_reset.py new file mode 100644 index 0000000..4e960eb --- /dev/null +++ b/backend/app/tasks/usage_reset.py @@ -0,0 +1,71 @@ +from datetime import date, datetime, timezone + +import structlog +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from app.celery_app import celery_app +from app.config import settings +from app.models.subscription import Subscription +from app.models.usage import UserUsage + +logger = structlog.get_logger() + + +def _calendar_month(now: datetime) -> tuple[date, date]: + start = date(now.year, now.month, 1) + end = date(now.year + 1, 1, 1) if now.month == 12 else date(now.year, now.month + 1, 1) + return start, end + + +@celery_app.task(name="reconcile_usage_periods") +def reconcile_usage_periods(): + """Daily reconciliation for the per-period `user_usage` table. + + Usage rows are created lazily on first increment, so this task isn't a + zeroing cron — it (1) pre-creates the next period's row on billing + rollover so dashboards don't show a transient empty state, and (2) flags + active paid subscriptions whose `current_period_end` has passed with no + renewal webhook received, which would otherwise silently strand a user + on stale usage limits until the next webhook retry. + """ + engine = create_engine(settings.DATABASE_URL_SYNC) + session = Session(engine) + try: + now = datetime.now(timezone.utc) + subs = session.query(Subscription).filter(Subscription.plan_id.isnot(None)).all() + + for sub in subs: + if ( + sub.plan_id + and sub.plan_id != "free" + and sub.subscription_status == "active" + and sub.current_period_end + and sub.current_period_end < now + ): + logger.warning( + "usage_reset.missed_renewal_webhook", + subscription_id=str(sub.id), + user_id=str(sub.user_id), + current_period_end=sub.current_period_end.isoformat(), + ) + continue + + if sub.current_period_start and sub.current_period_end and sub.subscription_status == "active": + period_start = sub.current_period_start.date() + period_end = sub.current_period_end.date() + else: + period_start, period_end = _calendar_month(now) + + existing = ( + session.query(UserUsage) + .filter(UserUsage.user_id == sub.user_id, UserUsage.period_start == period_start) + .first() + ) + if not existing: + session.add(UserUsage(user_id=sub.user_id, period_start=period_start, period_end=period_end)) + + session.commit() + finally: + session.close() + engine.dispose() diff --git a/backend/scripts/promote_admin.py b/backend/scripts/promote_admin.py new file mode 100644 index 0000000..cb6955f --- /dev/null +++ b/backend/scripts/promote_admin.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +""" +Promote a user to admin (role='admin' in user_roles). There is no in-app UI +for this — it's the interim mechanism until/unless an admin-invite flow is built. + +Usage: + docker exec jobforgex-backend-1 python /app/scripts/promote_admin.py user@example.com + +The user must log out and log back in afterward — the JWT's `role` claim is +baked in at login time and won't reflect the DB change until a fresh token +is issued. +""" + +import os +import sys + +import psycopg2 + +DB_HOST = os.getenv("DB_HOST", "postgres") +DB_PORT = os.getenv("DB_PORT", "5432") +DB_USER = os.getenv("DB_USER", "jobforgex") +DB_PASSWORD = os.getenv("DB_PASSWORD", "jobforgex") +DB_NAME = os.getenv("DB_NAME", "jobforgex") + + +def main(): + if len(sys.argv) != 2: + print("Usage: promote_admin.py ") + sys.exit(1) + + email = sys.argv[1].strip().lower() + conn = psycopg2.connect(host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, dbname=DB_NAME) + try: + cur = conn.cursor() + cur.execute("SELECT id FROM users WHERE lower(email) = %s", (email,)) + row = cur.fetchone() + if not row: + print(f"No user found with email {email!r}") + sys.exit(1) + user_id = row[0] + + cur.execute("SELECT id FROM user_roles WHERE user_id = %s", (user_id,)) + if cur.fetchone(): + cur.execute("UPDATE user_roles SET role = 'admin' WHERE user_id = %s", (user_id,)) + else: + cur.execute( + "INSERT INTO user_roles (id, user_id, role) VALUES (gen_random_uuid(), %s, 'admin')", (user_id,) + ) + conn.commit() + print(f"{email} is now an admin. They must log out and back in for it to take effect.") + finally: + conn.close() + + +if __name__ == "__main__": + main() diff --git a/database/migrations/001_init.sql b/database/migrations/001_init.sql index 5e6ebb4..388d8eb 100644 --- a/database/migrations/001_init.sql +++ b/database/migrations/001_init.sql @@ -443,15 +443,57 @@ CREATE TRIGGER trg_subscriptions_updated_at -- ============================================================================ INSERT INTO plans (id, name, monthly_price_usd, annual_price_usd, job_track_limit, cover_letter_limit, features, sort_order) VALUES ('free', 'Free', 0, 0, 2, 2, '["2 job tracks", "2 cover letters", "ATS checker", "Chrome extension"]', 0), - ('pro', 'Pro', 14.00, 99.00, 30, 30, '["30 job tracks", "30 cover letters", "Structured resume builder", "ATS checker", "Chrome extension", "Priority support"]', 1), - ('unlimited', 'Unlimited', 29.00, 199.00, NULL, NULL, '["Unlimited job tracks", "Unlimited cover letters", "Structured resume builder", "ATS checker", "Chrome extension", "Priority support"]', 2) + ('pro', 'Pro', 12.00, 99.00, 30, 30, '["30 job tracks", "30 cover letters", "Structured resume builder", "ATS checker", "Chrome extension", "Priority support"]', 1), + ('unlimited', 'Unlimited', 24.00, 199.00, NULL, NULL, '["Unlimited job tracks", "Unlimited cover letters", "Structured resume builder", "ATS checker", "Chrome extension", "Priority support"]', 2) ON CONFLICT (id) DO NOTHING; INSERT INTO geo_pricing (plan_id, country_code, currency, currency_symbol, monthly_price, annual_price, priority) VALUES ('pro', 'IN', 'INR', '₹', 599.00, 3999.00, 10), - ('pro', 'US', 'USD', '$', 14.00, 99.00, 5), - ('pro', 'DEFAULT', 'USD', '$', 14.00, 99.00, 0), + ('pro', 'US', 'USD', '$', 12.00, 99.00, 5), + ('pro', 'DEFAULT', 'USD', '$', 12.00, 99.00, 0), ('unlimited', 'IN', 'INR', '₹', 1299.00, 9999.00, 10), - ('unlimited', 'US', 'USD', '$', 29.00, 249.00, 5), - ('unlimited', 'DEFAULT', 'USD', '$', 29.00, 249.00, 0) + ('unlimited', 'US', 'USD', '$', 24.00, 199.00, 5), + ('unlimited', 'DEFAULT', 'USD', '$', 24.00, 199.00, 0) ON CONFLICT DO NOTHING; + +-- Self-heal: fix price mismatches on DBs that already applied the seed above with old values +UPDATE plans SET monthly_price_usd = 12.00 WHERE id = 'pro' AND monthly_price_usd <> 12.00; +UPDATE plans SET monthly_price_usd = 24.00 WHERE id = 'unlimited' AND monthly_price_usd <> 24.00; +UPDATE geo_pricing SET monthly_price = 12.00 WHERE plan_id = 'pro' AND country_code IN ('US', 'DEFAULT') AND monthly_price <> 12.00; +UPDATE geo_pricing SET monthly_price = 24.00, annual_price = 199.00 WHERE plan_id = 'unlimited' AND country_code IN ('US', 'DEFAULT') AND (monthly_price <> 24.00 OR annual_price <> 199.00); + +-- ============================================================================ +-- USER USAGE (per billing/calendar period counters for job tracks & cover letters) +-- ============================================================================ +CREATE TABLE user_usage ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + period_start DATE NOT NULL, + period_end DATE NOT NULL, + job_tracks_used INT NOT NULL DEFAULT 0, + cover_letters_used INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (user_id, period_start) +); +CREATE INDEX idx_user_usage_user ON user_usage (user_id); +CREATE INDEX idx_user_usage_period ON user_usage (period_start); + +CREATE TRIGGER trg_user_usage_updated_at + BEFORE UPDATE ON user_usage + FOR EACH ROW EXECUTE FUNCTION update_updated_at(); + +-- ============================================================================ +-- PAYMENT EVENTS: attach to user (a Workspace may not exist yet at webhook time) +-- ============================================================================ +ALTER TABLE payment_events ALTER COLUMN workspace_id DROP NOT NULL; +ALTER TABLE payment_events ADD COLUMN IF NOT EXISTS user_id UUID REFERENCES users(id) ON DELETE SET NULL; +CREATE INDEX IF NOT EXISTS idx_payment_events_user ON payment_events (user_id); + +-- ============================================================================ +-- SUBSCRIPTIONS: audit columns for billing history / webhook event tracking +-- ============================================================================ +ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS last_payment_status TEXT; +ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS last_payment_at TIMESTAMPTZ; +ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS paused_at TIMESTAMPTZ; +ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS cancelled_at TIMESTAMPTZ; diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 86e88d6..83278cd 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -13,6 +13,8 @@ import BillingPage from "@/pages/billing"; import SettingsPage from "@/pages/settings"; import LandingPage from "@/pages/landing"; import PrivacyPage from "@/pages/privacy"; +import AdminOverviewPage from "@/pages/admin/overview"; +import AdminSubscriptionsPage from "@/pages/admin/subscriptions"; function ProtectedRoute({ children }: { children: React.ReactNode }) { const { isAuthenticated, isLoading } = useAuth(); @@ -37,6 +39,15 @@ function ProtectedRoute({ children }: { children: React.ReactNode }) { ); } +function AdminRoute({ children }: { children: React.ReactNode }) { + const { user } = useAuth(); + return ( + + {user?.role === "admin" ? children : } + + ); +} + export default function App() { return ( <> @@ -53,6 +64,8 @@ export default function App() { } /> } /> } /> + } /> + } /> } /> diff --git a/frontend/src/api/admin.ts b/frontend/src/api/admin.ts new file mode 100644 index 0000000..a029317 --- /dev/null +++ b/frontend/src/api/admin.ts @@ -0,0 +1,61 @@ +import apiClient from "./client"; + +export interface PlanDistribution { + plan_id: string; + plan_name: string; + count: number; +} + +export interface AdminOverview { + total_users: number; + active_subscribers: number; + cancelled_subscribers: number; + suspended_subscribers: number; + free_users: number; + plan_distribution: PlanDistribution[]; + mrr_usd: number; +} + +export interface RevenuePoint { + date: string; + amount_usd: number; + event_count: number; +} + +export interface AdminSubscription { + id: string; + user_id: string; + email: string; + full_name: string | null; + plan: string; + subscription_status: string | null; + billing_cycle: string | null; + current_period_end: string | null; + cancel_at_period_end: boolean; + suspended: boolean; + created_at: string; +} + +export interface AdminUsageRow { + user_id: string; + email: string; + plan: string; + job_tracks_used: number; + cover_letters_used: number; + period_start: string; +} + +export const adminApi = { + getOverview: () => apiClient.get("/api/v1/admin/stats/overview").then((r) => r.data), + + getRevenue: (days = 30) => + apiClient.get("/api/v1/admin/stats/revenue", { params: { days } }).then((r) => r.data), + + listSubscriptions: (limit = 50, offset = 0) => + apiClient + .get("/api/v1/admin/subscriptions", { params: { limit, offset } }) + .then((r) => r.data), + + listUsage: (limit = 50) => + apiClient.get("/api/v1/admin/usage", { params: { limit } }).then((r) => r.data), +}; diff --git a/frontend/src/api/billing.ts b/frontend/src/api/billing.ts index 34cbf9e..4f44856 100644 --- a/frontend/src/api/billing.ts +++ b/frontend/src/api/billing.ts @@ -7,6 +7,7 @@ export interface BillingStatus { trial_limit: number; has_pro: boolean; current_period_end: string | null; + cancel_at_period_end: boolean; } export interface Subscription { @@ -16,9 +17,15 @@ export interface Subscription { subscription_status: string | null; razorpay_subscription_id: string | null; billing_cycle: string | null; + current_period_start: string | null; current_period_end: string | null; + cancel_at_period_end: boolean; trial_ends_at: string | null; suspended: boolean; + last_payment_status: string | null; + last_payment_at: string | null; + paused_at: string | null; + cancelled_at: string | null; } export interface Pricing { @@ -30,10 +37,35 @@ export interface Pricing { annual_price: number; } +export interface UsageMetric { + used: number; + limit: number | null; + remaining: number | null; + unlimited: boolean; +} + +export interface Usage { + plan: string; + period_start: string; + period_end: string; + job_tracks: UsageMetric; + cover_letters: UsageMetric; +} + +export interface BillingHistoryItem { + id: string; + event_type: string; + created_at: string; + summary: string; +} + export const billingApi = { getStatus: () => apiClient.get("/api/v1/billing/status").then((r) => r.data), + getUsage: () => + apiClient.get("/api/v1/billing/usage").then((r) => r.data), + createSubscription: (data: { plan_id?: string; billing_cycle?: string; @@ -41,9 +73,20 @@ export const billingApi = { trial?: boolean; }) => apiClient.post("/api/v1/billing/subscription/create", data).then((r) => r.data), + cancelSubscription: (atPeriodEnd: boolean = true) => + apiClient + .post("/api/v1/billing/subscription/cancel", { at_period_end: atPeriodEnd }) + .then((r) => r.data), + + reactivateSubscription: () => + apiClient.post("/api/v1/billing/subscription/reactivate").then((r) => r.data), + getSubscription: () => apiClient.get("/api/v1/billing/subscription").then((r) => r.data), + getBillingHistory: () => + apiClient.get("/api/v1/billing/subscription/history").then((r) => r.data), + getPricing: (countryCode?: string) => apiClient .get("/api/v1/billing/pricing", { params: { country_code: countryCode || "DEFAULT" } }) diff --git a/frontend/src/components/app-shell.tsx b/frontend/src/components/app-shell.tsx index 8c7233f..ff331a1 100644 --- a/frontend/src/components/app-shell.tsx +++ b/frontend/src/components/app-shell.tsx @@ -1,13 +1,17 @@ import { Link, useLocation, useNavigate } from "react-router-dom"; -import { ListTodo, Wand2, FileText, ClipboardCheck, CreditCard, Settings, Sparkles, LogOut } from "lucide-react"; +import { ListTodo, Wand2, FileText, ClipboardCheck, CreditCard, Settings, Sparkles, LogOut, ShieldCheck } from "lucide-react"; import { toast } from "sonner"; -import { useQueryClient } from "@tanstack/react-query"; +import { useQueryClient, useQuery } from "@tanstack/react-query"; import logoImg from "@/assets/logo.png"; import { cn } from "@/lib/utils"; import { ThemeToggle } from "@/components/theme-toggle"; +import { Badge } from "@/components/ui/badge"; import { useAuth } from "@/context/auth-context"; import { useWorkspace } from "@/hooks/use-workspace"; +import { billingApi } from "@/api/billing"; + +const PLAN_LABELS: Record = { free: "Free", pro: "Pro", unlimited: "Unlimited" }; const NAV = [ { to: "/jobs", label: "Jobs", icon: ListTodo }, @@ -22,8 +26,10 @@ export function AppShell({ children }: { children: React.ReactNode }) { const location = useLocation(); const navigate = useNavigate(); const qc = useQueryClient(); - const { logout } = useAuth(); + const { logout, user } = useAuth(); const { data: ws } = useWorkspace(); + const { data: billing } = useQuery({ queryKey: ["billing"], queryFn: () => billingApi.getStatus() }); + const nav = user?.role === "admin" ? [...NAV, { to: "/admin", label: "Admin", icon: ShieldCheck }] : NAV; async function signOut() { await qc.cancelQueries(); @@ -44,7 +50,7 @@ export function AppShell({ children }: { children: React.ReactNode }) { JobForge