# Indie Status Page — SaaS Enhancement Plan > Transform the single-tenant status page into a multi-tenant SaaS product with > user accounts, org-level status pages, feature tiers, and Stripe payment links. --- ## 1. Current State Analysis ### 1.1 Architecture Overview | Aspect | Detail | |---|---| | **Framework** | FastAPI 0.110+ with async SQLAlchemy 2.0 + aiosqlite | | **Database** | SQLite (`data/statuspage.db`), auto-created via `init_db()` — no Alembic migrations yet | | **Deployment** | Port 8765, nginx reverse proxy at `korpo.pro/status/`, systemd service `indie-status-page.service` | | **Templating** | Jinja2 (5 HTML templates) | | **Scheduling** | APScheduler (60s interval uptime checks) | ### 1.2 Database Models (8 total) File: `app/models/models.py` | Model | Table | Key Fields | Relationships | |---|---|---|---| | `Service` | `services` | id, name, slug (unique), description, group_name, position, is_visible | → incidents, → monitors | | `Incident` | `incidents` | id, service_id (FK), title, status, severity, started_at, resolved_at | → service, → updates, → notifications | | `IncidentUpdate` | `incident_updates` | id, incident_id (FK), status, body, created_at | → incident | | `Monitor` | `monitors` | id, service_id (FK), url, method, expected_status, timeout_seconds, interval_seconds, is_active | → service, → results | | `MonitorResult` | `monitor_results` | id, monitor_id (FK), status, response_time_ms, status_code, error_message, checked_at | → monitor | | `Subscriber` | `subscribers` | id, email (unique), is_confirmed, confirm_token | → notifications | | `NotificationLog` | `notification_logs` | id, incident_id (FK), subscriber_id (FK), channel, status | → incident, → subscriber | | `SiteSetting` | `site_settings` | id, key (unique), value | — | ### 1.3 API Routes (25 endpoints) File: `app/api/router.py` — assembles 5 sub-routers under `/api/v1`: | Router | File | Prefix | Endpoints | |---|---|---|---| | **Services** | `app/api/services.py` | `/services` | GET /, POST /, GET /{id}, PATCH /{id}, DELETE /{id} | | **Incidents** | `app/api/incidents.py` | `/incidents` | GET /, POST /, GET /{id}, PATCH /{id}, DELETE /{id}, POST /{id}/updates | | **Monitors** | `app/api/monitors.py` | `/monitors` | GET /, POST /, GET /{id}, PATCH /{id}, DELETE /{id}, POST /{id}/check | | **Subscribers** | `app/api/subscribers.py` | `/subscribers` | GET /, POST /, DELETE /{id}, POST /{id}/confirm | | **Settings** | `app/api/settings.py` | `/settings` | GET /, PATCH / | ### 1.4 Authentication File: `app/dependencies.py` - **Single shared API key**: `X-API-Key` header validated against `settings.admin_api_key` - No user accounts, no sessions, no JWT tokens - All write endpoints require API key; read endpoints are public ### 1.5 Frontend Pages File: `app/main.py` + `app/templates/` | Route | Template | Purpose | |---|---|---| | `GET /` | `status.html` | Public status page (all visible services + recent incidents) | | `GET /incident/{id}` | `incident.html` | Incident detail with update timeline | | `GET /subscribe` | `subscribe.html` | Email subscription form | | `GET /confirm` | `confirm.html` | Subscription confirmation | | `GET /health` | JSON response | Health check for container orchestration | ### 1.6 Background Services | Service | File | Description | |---|---|---| | **Uptime Checker** | `app/services/uptime.py` | HTTP health checks via httpx, stores `MonitorResult` | | **Scheduler** | `app/services/scheduler.py` | APScheduler runs `_run_monitor_checks()` every 60s | | **Notifier** | `app/services/notifier.py` | Email (SMTP) + webhook dispatch for incident updates | ### 1.7 Tests 6 passing tests across 3 files: - `tests/test_health.py` — 1 test (health check) - `tests/test_api_services.py` — 7 tests (full CRUD + 404 + auth) - `tests/test_api_incidents.py` — 5 tests (CRUD + updates + delete) ### 1.8 Key Gaps for SaaS | Gap | Impact | |---|---| | **No user accounts** | Cannot identify who owns what data | | **No multi-tenancy** | All orgs share the same data; no isolation | | **No feature limits** | Any user can create unlimited services/pages | | **No payment integration** | No billing, no subscription management | | **SQLite-only** | Won't handle concurrent multi-tenant writes well | | **No Alembic migrations** | Schema changes require manual DB recreation | | **Single-site config** | `SiteSetting` and `config.py` are global, not per-org | --- ## 2. Multi-Tenancy Design ### 2.1 Strategy: Shared Database, Tenant-ID Columns Use a **shared database with tenant-ID discriminator** approach. Every tenant-scoped table gets an `organization_id` foreign key. All queries filter by the current tenant. This is the simplest path that works with SQLite → PostgreSQL migration. ### 2.2 New Models #### `User` — Individual who can log in ```python # app/models/models.py — NEW class User(Base): __tablename__ = "users" id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid_str) email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True) password_hash: Mapped[str] = mapped_column(String(255), nullable=False) display_name: Mapped[str | None] = mapped_column(String(100), nullable=True) is_email_verified: Mapped[bool] = mapped_column(Boolean, default=False) email_verify_token: Mapped[str | None] = mapped_column(String(100), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) updated_at: Mapped[datetime] = mapped_column( DateTime, default=datetime.utcnow, onupdate=datetime.utcnow ) memberships: Mapped[list["OrganizationMember"]] = relationship(back_populates="user") ``` #### `Organization` — The tenant; owns status pages ```python class Organization(Base): __tablename__ = "organizations" id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid_str) slug: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True) name: Mapped[str] = mapped_column(String(100), nullable=False) tier: Mapped[str] = mapped_column(String(20), nullable=False, default="free") # "free" | "pro" | "team" stripe_customer_id: Mapped[str | None] = mapped_column(String(100), nullable=True) custom_domain: Mapped[str | None] = mapped_column(String(255), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) updated_at: Mapped[datetime] = mapped_column( DateTime, default=datetime.utcnow, onupdate=datetime.utcnow ) members: Mapped[list["OrganizationMember"]] = relationship(back_populates="organization") status_pages: Mapped[list["StatusPage"]] = relationship(back_populates="organization") ``` #### `OrganizationMember` — Joins users to orgs with roles ```python class OrganizationMember(Base): __tablename__ = "organization_members" id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid_str) organization_id: Mapped[str] = mapped_column( String(36), ForeignKey("organizations.id"), nullable=False, index=True ) user_id: Mapped[str] = mapped_column( String(36), ForeignKey("users.id"), nullable=False, index=True ) role: Mapped[str] = mapped_column(String(20), nullable=False, default="member") # "owner" | "admin" | "member" joined_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) organization: Mapped["Organization"] = relationship(back_populates="members") user: Mapped["User"] = relationship(back_populates="memberships") __table_args__ = ( # Prevent duplicate memberships {"sqlite_autoincrement": True}, ) ``` #### `StatusPage` — Replaces the single global status view ```python class StatusPage(Base): __tablename__ = "status_pages" id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid_str) organization_id: Mapped[str] = mapped_column( String(36), ForeignKey("organizations.id"), nullable=False, index=True ) slug: Mapped[str] = mapped_column(String(50), nullable=False, index=True) title: Mapped[str] = mapped_column(String(100), nullable=False) subdomain: Mapped[str | None] = mapped_column(String(100), nullable=True) custom_domain: Mapped[str | None] = mapped_column(String(255), nullable=True) is_public: Mapped[bool] = mapped_column(Boolean, default=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) updated_at: Mapped[datetime] = mapped_column( DateTime, default=datetime.utcnow, onupdate=datetime.utcnow ) organization: Mapped["Organization"] = relationship(back_populates="status_pages") __table_args__ = ( # Unique slug within organization UniqueConstraint("organization_id", "slug", name="uq_status_page_org_slug"), ) ``` ### 2.3 Tenant-ID on Existing Models Every existing model (except `SiteSetting`) gains an `organization_id` FK: ```python # Add to Service: organization_id: Mapped[str] = mapped_column( String(36), ForeignKey("organizations.id"), nullable=False, index=True ) # Add to Subscriber: organization_id: Mapped[str] = mapped_column( String(36), ForeignKey("organizations.id"), nullable=False, index=True ) # SiteSetting gets organization_id (nullable for global settings): organization_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("organizations.id"), nullable=True, index=True ) ``` `Incident`, `IncidentUpdate`, `Monitor`, `MonitorResult`, and `NotificationLog` are implicitly scoped through their parent `Service` → no direct `organization_id` needed if all queries join through `Service`. However, for query performance, add a denormalized `organization_id` on `Incident` and `Monitor`: ```python # Add to Incident: organization_id: Mapped[str] = mapped_column( String(36), ForeignKey("organizations.id"), nullable=False, index=True ) # Add to Monitor: organization_id: Mapped[str] = mapped_column( String(36), ForeignKey("organizations.id"), nullable=False, index=True ) ``` ### 2.4 Auth Flow Replace `verify_api_key` with JWT-based user auth: **New file: `app/api/auth.py`** ```python from datetime import datetime, timedelta from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import jwt, JWTError from pydantic import BaseModel, EmailStr from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.dependencies import get_db from app.models.models import User, OrganizationMember, Organization router = APIRouter() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") class LoginRequest(BaseModel): email: EmailStr password: str class RegisterRequest(BaseModel): email: EmailStr password: str display_name: str | None = None org_name: str # Name for auto-created org class TokenResponse(BaseModel): access_token: str token_type: str = "bearer" def create_access_token(user_id: str, exp_hours: int = 72) -> str: payload = { "sub": user_id, "exp": datetime.utcnow() + timedelta(hours=exp_hours), } return jwt.encode(payload, settings.secret_key, algorithm="HS256") async def get_current_user( token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db), ) -> User: """Dependency: extract and validate current user from JWT.""" try: payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"]) user_id = payload.get("sub") except JWTError: raise HTTPException(status_code=401, detail="Invalid token") result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=401, detail="User not found") return user @router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED) async def register(data: RegisterRequest, db: AsyncSession = Depends(get_db)): """Register a new user + auto-create a personal organization.""" # Check if email exists existing = await db.execute(select(User).where(User.email == data.email)) if existing.scalar_one_or_none(): raise HTTPException(status_code=409, detail="Email already registered") from passlib.context import CryptContext pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto") user = User( email=data.email, password_hash=pwd_ctx.hash(data.password), display_name=data.display_name, ) db.add(user) await db.flush() # Auto-create personal org org_slug = data.email.split("@")[0].lower().replace(".", "-") org = Organization(name=data.org_name, slug=org_slug, tier="free") db.add(org) await db.flush() membership = OrganizationMember( organization_id=org.id, user_id=user.id, role="owner" ) db.add(membership) await db.flush() token = create_access_token(user.id) return TokenResponse(access_token=token) @router.post("/login", response_model=TokenResponse) async def login(data: LoginRequest, db: AsyncSession = Depends(get_db)): result = await db.execute(select(User).where(User.email == data.email)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=401, detail="Invalid credentials") from passlib.context import CryptContext pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto") if not pwd_ctx.verify(data.password, user.password_hash): raise HTTPException(status_code=401, detail="Invalid credentials") token = create_access_token(user.id) return TokenResponse(access_token=token) ``` ### 2.5 Tenant-Scoped Query Pattern Replace all `select(Service)` queries with tenant-filtered variants: ```python # Before (app/api/services.py line 51): result = await db.execute(select(Service).order_by(Service.position, Service.name)) # After: async def get_current_org( user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> Organization: """Get the user's first org (simplified; add org-switching later).""" result = await db.execute( select(OrganizationMember).where(OrganizationMember.user_id == user.id) ) membership = result.scalars().first() if not membership: raise HTTPException(status_code=403, detail="No organization") org_result = await db.execute( select(Organization).where(Organization.id == membership.organization_id) ) return org_result.scalar_one() # In endpoints: org = Depends(get_current_org) result = await db.execute( select(Service) .where(Service.organization_id == org.id) .order_by(Service.position, Service.name) ) ``` ### 2.6 Multi-Page Public Routes Current `GET /` serves one global page. New design: | Route | Purpose | |---|---| | `GET /` | Landing / marketing page | | `GET /p/{org_slug}` | Org's default status page | | `GET /p/{org_slug}/{page_slug}` | Specific status page | | `GET /p/{org_slug}/incident/{id}` | Incident detail within org context | | Custom domain | Resolves to `StatusPage.custom_domain` → renders that page | ```python # app/main.py — NEW route @app.get("/p/{org_slug}/{page_slug}") async def public_status_page(request: Request, org_slug: str, page_slug: str): from sqlalchemy import select from app.database import async_session_factory from app.models.models import Organization, StatusPage, Service, Incident async with async_session_factory() as db: # Resolve org + page org_result = await db.execute( select(Organization).where(Organization.slug == org_slug) ) org = org_result.scalar_one_or_none() if not org: raise HTTPException(status_code=404) page_result = await db.execute( select(StatusPage).where( StatusPage.organization_id == org.id, StatusPage.slug == page_slug, ) ) page = page_result.scalar_one_or_none() if not page: raise HTTPException(status_code=404) # Get services for this org result = await db.execute( select(Service) .where(Service.organization_id == org.id, Service.is_visible == True) .order_by(Service.position, Service.name) ) services = result.scalars().all() # ... render template with org + page context ``` ### 2.7 Custom Domain Resolution Add middleware to resolve custom domains: ```python # app/middleware.py — NEW from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from sqlalchemy import select from app.database import async_session_factory from app.models.models import StatusPage, Organization class CustomDomainMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): host = request.headers.get("host", "").split(":")[0] # Skip known app domains if host in ("korpo.pro", "localhost", "127.0.0.1"): return await call_next(request) async with async_session_factory() as db: result = await db.execute( select(StatusPage).where(StatusPage.custom_domain == host) ) page = result.scalar_one_or_none() if page: # Store page info in request state for downstream use request.state.custom_domain_page = page return await call_next(request) ``` --- ## 3. Feature Tiers ### 3.1 Tier Definitions | Feature | Free | Pro ($9/mo) | Team ($29/mo) | |---|---|---|---| | **Status pages** | 1 | 5 | Unlimited | | **Services per page** | 5 | 50 | Unlimited | | **Monitors per service** | 1 | 5 | Unlimited | | **Subscribers** | 25 | 500 | Unlimited | | **Uptime check interval** | 5 min | 1 min | 30 sec | | **Custom domain** | ❌ | ✅ | ✅ | | **Custom branding/CSS** | ❌ | ✅ | ✅ | | **Team members** | 1 | 3 | Unlimited | | **Incident history** | 30 days | 1 year | Unlimited | | **Webhook notifications** | ❌ | ✅ | ✅ | | **API access** | ❌ | ✅ | ✅ | | **Email notifications** | ✅ | ✅ | ✅ | | **SLA badge widget** | ❌ | ✅ | ✅ | | **Status page password protection** | ❌ | ❌ | ✅ | ### 3.2 Enforcement Layer Create a tier-limits module that checks against the org's tier before allowing writes: **New file: `app/services/tier_limits.py`** ```python from fastapi import HTTPException from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from app.models.models import ( Organization, Service, Monitor, Subscriber, StatusPage, OrganizationMember ) TIER_LIMITS = { "free": { "status_pages": 1, "services_per_page": 5, "monitors_per_service": 1, "subscribers": 25, "members": 1, "check_interval_min": 5, "custom_domain": False, "webhooks": False, "api_access": False, }, "pro": { "status_pages": 5, "services_per_page": 50, "monitors_per_service": 5, "subscribers": 500, "members": 3, "check_interval_min": 1, "custom_domain": True, "webhooks": True, "api_access": True, }, "team": { "status_pages": -1, # unlimited "services_per_page": -1, "monitors_per_service": -1, "subscribers": -1, "members": -1, "check_interval_min": 0, # 30 sec "custom_domain": True, "webhooks": True, "api_access": True, }, } def get_limit(org: Organization, feature: str): """Return the limit for a feature given the org's tier.""" tier = org.tier or "free" return TIER_LIMITS.get(tier, TIER_LIMITS["free"]).get(feature) async def enforce_limit( db: AsyncSession, org: Organization, feature: str, current_count: int ) -> None: """Raise 403 if the current count meets or exceeds the tier limit.""" limit = get_limit(org, feature) if limit == -1: # unlimited return if limit is False: # feature not available raise HTTPException( status_code=403, detail=f"Feature '{feature}' requires a plan upgrade", ) if current_count >= limit: raise HTTPException( status_code=403, detail=f"Tier limit reached for '{feature}' ({limit}). Upgrade your plan.", ) # Concrete helpers: async def check_status_page_limit(db: AsyncSession, org: Organization): count = await db.execute( select(func.count(StatusPage.id)).where(StatusPage.organization_id == org.id) ) await enforce_limit(db, org, "status_pages", count.scalar() or 0) async def check_service_limit(db: AsyncSession, org: Organization, page_id: str): count = await db.execute( select(func.count(Service.id)).where( Service.organization_id == org.id, Service.status_page_id == page_id, ) ) await enforce_limit(db, org, "services_per_page", count.scalar() or 0) async def check_monitor_limit(db: AsyncSession, org: Organization, service_id: str): count = await db.execute( select(func.count(Monitor.id)).where(Monitor.service_id == service_id) ) await enforce_limit(db, org, "monitors_per_service", count.scalar() or 0) async def check_subscriber_limit(db: AsyncSession, org: Organization): count = await db.execute( select(func.count(Subscriber.id)).where(Subscriber.organization_id == org.id) ) await enforce_limit(db, org, "subscribers", count.scalar() or 0) async def check_member_limit(db: AsyncSession, org: Organization): count = await db.execute( select(func.count(OrganizationMember.id)).where( OrganizationMember.organization_id == org.id ) ) await enforce_limit(db, org, "members", count.scalar() or 0) ``` ### 3.3 Usage in Endpoints ```python # app/api/services.py — modify create_service @router.post("/", status_code=status.HTTP_201_CREATED) async def create_service( data: ServiceCreate, org: Organization = Depends(get_current_org), user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): from app.services.tier_limits import check_service_limit await check_service_limit(db, org, data.status_page_id) # ENFORCE LIMIT service = Service( organization_id=org.id, name=data.name, slug=data.slug, # ... ) db.add(service) # ... ``` --- ## 4. Payment Integration — Stripe Checkout Links ### 4.1 Approach: No-Code Payment Links Use **Stripe Checkout Links** (Payment Links) — pre-built hosted checkout pages that require zero server-side Stripe integration. When a user clicks "Upgrade", they're redirected to a Stripe-hosted checkout page. After payment, Stripe redirects back with a `?session_id` parameter. A webhook marks the org as upgraded. ### 4.2 Setup in Stripe Dashboard 1. Create 2 Products: "Pro Plan" ($9/mo) and "Team Plan" ($29/mo) 2. Create recurring prices for each 3. Generate **Payment Links** for each price: - Pro: `https://buy.stripe.com/xxxx_pro_9_monthly` - Team: `https://buy.stripe.com/xxxx_team_29_monthly` 4. Set success URL: `https://korpo.pro/api/v1/billing/success?session_id={CHECKOUT_SESSION_ID}` 5. Set cancel URL: `https://korpo.pro/billing?canceled=1` ### 4.3 Config — Add to `app/config.py` ```python class Settings(BaseSettings): # ... existing fields ... # Stripe stripe_pro_checkout_url: str = "https://buy.stripe.com/xxxx_pro_9_monthly" stripe_team_checkout_url: str = "https://buy.stripe.com/xxxx_team_29_monthly" stripe_webhook_secret: str = "whsec_xxxx" ``` ### 4.4 Checkout Redirect Endpoint **New file: `app/api/billing.py`** ```python from fastapi import APIRouter, Depends, HTTPException, Request from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.dependencies import get_db from app.api.auth import get_current_user from app.models.models import User, Organization, OrganizationMember router = APIRouter() @router.get("/checkout/{tier}") async def create_checkout( tier: str, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Redirect user to Stripe Checkout Link for the selected tier.""" if tier not in ("pro", "team"): raise HTTPException(status_code=400, detail="Invalid tier") # Get user's org result = await db.execute( select(OrganizationMember).where(OrganizationMember.user_id == user.id) ) membership = result.scalars().first() if not membership: raise HTTPException(status_code=403, detail="No organization") # Select checkout URL based on tier checkout_url = ( settings.stripe_pro_checkout_url if tier == "pro" else settings.stripe_team_checkout_url ) # Append metadata so Stripe webhook knows which org to upgrade # Stripe Payment Links support client_reference_id via URL param: from urllib.parse import urlencode params = urlencode({ "client_reference_id": membership.organization_id, "prefilled_email": user.email, }) return {"redirect_url": f"{checkout_url}?{params}"} ``` ### 4.5 Stripe Webhook Handler This is the **only** server-side Stripe code needed. It receives events when payments succeed or subscriptions change. ```python # app/api/billing.py — continued import stripe from fastapi import Header @router.post("/webhook") async def stripe_webhook(request: Request, db: AsyncSession = Depends(get_db)): """Handle Stripe webhook events for subscription changes.""" body = await request.body() sig = request.headers.get("stripe-signature", "") try: event = stripe.Webhook.construct_event( body, sig, settings.stripe_webhook_secret ) except Exception: raise HTTPException(status_code=400, detail="Invalid signature") if event["type"] == "checkout.session.completed": session = event["data"]["object"] org_id = session.get("client_reference_id") customer_id = session.get("customer") if org_id: result = await db.execute( select(Organization).where(Organization.id == org_id) ) org = result.scalar_one_or_none() if org: # Determine tier from the line items amount = session.get("amount_total", 0) if amount >= 2900: # $29+ org.tier = "team" elif amount >= 900: # $9+ org.tier = "pro" org.stripe_customer_id = customer_id await db.flush() elif event["type"] == "customer.subscription.deleted": # Downgrade to free on cancellation customer_id = event["data"]["object"]["customer"] result = await db.execute( select(Organization).where(Organization.stripe_customer_id == customer_id) ) org = result.scalar_one_or_none() if org: org.tier = "free" await db.flush() return {"status": "ok"} ``` ### 4.6 Success Callback ```python @router.get("/success") async def billing_success(session_id: str, db: AsyncSession = Depends(get_db)): """Stripe redirects here after successful checkout.""" # In production, call stripe.checkout.sessions.retrieve(session_id) to verify. # For MVP, webhook already processed it — just show a success page. return {"message": "Subscription activated! Your plan has been upgraded."} ``` ### 4.7 Billing Page Template ```html {% extends "base.html" %} {% block title %}Billing{% endblock %} {% block content %}

Choose Your Plan

Free

$0/mo

  • 1 status page
  • 5 services
  • 25 subscribers
  • Email notifications
{% if org.tier == 'free' %}

✅ Current plan

{% else %}Downgrade{% endif %}

Pro

$9/mo

  • 5 status pages
  • 50 services per page
  • Custom domain
  • Webhook notifications
  • API access
{% if org.tier == 'pro' %}

✅ Current plan

{% else %}Upgrade to Pro{% endif %}

Team

$29/mo

  • Unlimited pages & services
  • Unlimited team members
  • 30-second check intervals
  • Password-protected pages
{% if org.tier == 'team' %}

✅ Current plan

{% else %}Upgrade to Team{% endif %}
{% endblock %} ``` --- ## 5. Database Migration Steps ### 5.1 Set Up Alembic (Currently a Placeholder) The `alembic.ini` file is a stub. Full setup required: ```bash cd ~/wealth-engine/indie-status-page alembic init migrations # already has dir structure, but re-init config ``` Edit `alembic.ini`: ```ini sqlalchemy.url = sqlite+aiosqlite:///./data/statuspage.db ``` Edit `migrations/env.py` to import `Base` and all models: ```python from app.database import Base from app.models.models import * # ensure all models imported target_metadata = Base.metadata ``` ### 5.2 Migration 001: Add User + Organization Tables **Create a baseline migration for existing schema, then add new tables.** ```bash alembic revision --autogenerate -m "001_add_users_and_organizations" ``` ```python # migrations/versions/001_add_users_and_organizations.py def upgrade() -> None: # New tables — no existing data affected op.create_table( "users", sa.Column("id", sa.String(36), primary_key=True), sa.Column("email", sa.String(255), unique=True, nullable=False), sa.Column("password_hash", sa.String(255), nullable=False), sa.Column("display_name", sa.String(100)), sa.Column("is_email_verified", sa.Boolean, default=False), sa.Column("email_verify_token", sa.String(100)), sa.Column("created_at", sa.DateTime), sa.Column("updated_at", sa.DateTime), ) op.create_index("ix_users_email", "users", ["email"]) op.create_table( "organizations", sa.Column("id", sa.String(36), primary_key=True), sa.Column("slug", sa.String(50), unique=True, nullable=False), sa.Column("name", sa.String(100), nullable=False), sa.Column("tier", sa.String(20), nullable=False, server_default="free"), sa.Column("stripe_customer_id", sa.String(100)), sa.Column("custom_domain", sa.String(255)), sa.Column("created_at", sa.DateTime), sa.Column("updated_at", sa.DateTime), ) op.create_index("ix_organizations_slug", "organizations", ["slug"]) op.create_table( "organization_members", sa.Column("id", sa.String(36), primary_key=True), sa.Column("organization_id", sa.String(36), sa.ForeignKey("organizations.id"), nullable=False), sa.Column("user_id", sa.String(36), sa.ForeignKey("users.id"), nullable=False), sa.Column("role", sa.String(20), nullable=False, server_default="member"), sa.Column("joined_at", sa.DateTime), ) op.create_index("ix_org_members_org", "organization_members", ["organization_id"]) op.create_index("ix_org_members_user", "organization_members", ["user_id"]) op.create_table( "status_pages", sa.Column("id", sa.String(36), primary_key=True), sa.Column("organization_id", sa.String(36), sa.ForeignKey("organizations.id"), nullable=False), sa.Column("slug", sa.String(50), nullable=False), sa.Column("title", sa.String(100), nullable=False), sa.Column("subdomain", sa.String(100)), sa.Column("custom_domain", sa.String(255)), sa.Column("is_public", sa.Boolean, default=True), sa.Column("created_at", sa.DateTime), sa.Column("updated_at", sa.DateTime), sa.UniqueConstraint("organization_id", "slug", name="uq_status_page_org_slug"), ) op.create_index("ix_status_pages_org", "status_pages", ["organization_id"]) def downgrade() -> None: op.drop_table("status_pages") op.drop_table("organization_members") op.drop_table("organizations") op.drop_table("users") ``` ### 5.3 Migration 002: Add organization_id to Existing Tables ```bash alembic revision -m "002_add_organization_id_to_existing_tables" ``` ```python # migrations/versions/002_add_organization_id_to_existing_tables.py def upgrade() -> None: # Add nullable org_id columns first op.add_column("services", sa.Column( "organization_id", sa.String(36), sa.ForeignKey("organizations.id"), nullable=True )) op.add_column("incidents", sa.Column( "organization_id", sa.String(36), sa.ForeignKey("organizations.id"), nullable=True )) op.add_column("monitors", sa.Column( "organization_id", sa.String(36), sa.ForeignKey("organizations.id"), nullable=True )) op.add_column("subscribers", sa.Column( "organization_id", sa.String(36), sa.ForeignKey("organizations.id"), nullable=True )) op.add_column("site_settings", sa.Column( "organization_id", sa.String(36), sa.ForeignKey("organizations.id"), nullable=True )) # Add status_page_id column to services op.add_column("services", sa.Column( "status_page_id", sa.String(36), sa.ForeignKey("status_pages.id"), nullable=True )) # Create indexes for table in ("services", "incidents", "monitors", "subscribers", "site_settings"): op.create_index(f"ix_{table}_org", table, ["organization_id"]) op.create_index("ix_services_page", "services", ["status_page_id"]) def downgrade() -> None: for table in ("services", "incidents", "monitors", "subscribers", "site_settings"): op.drop_index(f"ix_{table}_org", table) op.drop_column(table, "organization_id") op.drop_index("ix_services_page", "services") op.drop_column("services", "status_page_id") ``` ### 5.4 Migration 003: Data Backfill — Migrate Existing Data to a Default Org ```bash alembic revision -m "003_backfill_organization_id" ``` ```python # migrations/versions/003_backfill_organization_id.py from uuid import uuid4 def upgrade() -> None: # 1. Create a default org for existing data default_org_id = str(uuid4()) op.execute(f""" INSERT INTO organizations (id, slug, name, tier, created_at, updated_at) VALUES ('{default_org_id}', 'default', 'Default Organization', 'free', datetime('now'), datetime('now')) """) # 2. Create a default status page default_page_id = str(uuid4()) op.execute(f""" INSERT INTO status_pages (id, organization_id, slug, title, is_public, created_at, updated_at) VALUES ('{default_page_id}', '{default_org_id}', 'default', 'Status Page', 1, datetime('now'), datetime('now')) """) # 3. Backfill organization_id on all existing rows for table in ("services", "incidents", "monitors", "subscribers", "site_settings"): op.execute(f""" UPDATE {table} SET organization_id = '{default_org_id}' WHERE organization_id IS NULL """) # 4. Backfill status_page_id on services op.execute(f""" UPDATE services SET status_page_id = '{default_page_id}' WHERE status_page_id IS NULL """) # 5. Now make organization_id NOT NULL # (SQLite doesn't support ALTER COLUMN, so recreate tables — # in PostgreSQL, this is just ALTER COLUMN SET NOT NULL) # For SQLite, we'll keep nullable and enforce at app level. # Document: enforce in model definition with nullable=False # after migrating to PostgreSQL. def downgrade() -> None: # Cannot reverse the backfill meaningfully pass ``` ### 5.5 Migration 004: SQLite → PostgreSQL (Production) For production multi-tenancy, migrate from SQLite to PostgreSQL: 1. Add `psycopg2-binary` + `asyncpg` to `pyproject.toml` dependencies 2. Change `DATABASE_URL` env var to `postgresql+asyncpg://...` 3. Use `pg_dump`/`pg_restore` or SQLAlchemy data migration script 4. Enable proper `ALTER COLUMN SET NOT NULL` on `organization_id` fields ```toml # pyproject.toml — add to dependencies "asyncpg>=0.29,<1.0", "psycopg2-binary>=2.9,<3.0", ``` ```python # app/config.py — update default database_url: str = "postgresql+asyncpg://statuspage:password@localhost/statuspage" ``` --- ## 6. Priority-Ordered Implementation Steps ### Phase 1: Foundation (Week 1–2) | # | Step | Files | Effort | |---|---|---|---| | 1 | **Set up Alembic properly** with all existing models as baseline | `alembic.ini`, `migrations/env.py`, `migrations/versions/000_baseline.py` | 2h | | 2 | **Add User model** + password hashing (passlib/bcrypt) | `app/models/models.py`, `pyproject.toml` (add `passlib[bcrypt]`, `python-jose`) | 2h | | 3 | **Add Organization + OrganizationMember + StatusPage models** | `app/models/models.py` | 2h | | 4 | **Create Alembic migration 001** (new tables only) | `migrations/versions/001_*.py` | 1h | | 5 | **Build auth endpoints** (register, login, JWT) | `app/api/auth.py` (new), `app/api/router.py` (add route) | 4h | | 6 | **Create `get_current_user` and `get_current_org` dependencies** | `app/dependencies.py` (extend) | 2h | | 7 | **Write auth tests** (register, login, token validation) | `tests/test_api_auth.py` (new) | 3h | ### Phase 2: Multi-Tenancy (Week 2–3) | # | Step | Files | Effort | |---|---|---|---| | 8 | **Add `organization_id` to existing models + migration 002** | `app/models/models.py`, `migrations/versions/002_*.py` | 3h | | 9 | **Data backfill migration 003** — create default org, assign all existing data | `migrations/versions/003_*.py` | 2h | | 10 | **Refactor all API endpoints** to filter by `organization_id` | `app/api/services.py`, `incidents.py`, `monitors.py`, `subscribers.py`, `settings.py` | 6h | | 11 | **Add `status_page_id` to Service** and update queries to scope by page | `app/models/models.py`, all API files | 2h | | 12 | **Build multi-page public routes** (`/p/{org_slug}/{page_slug}`) | `app/main.py`, `app/templates/status.html` (parametrize by org) | 4h | | 13 | **Build org management API** (create org, invite members, switch org) | `app/api/organizations.py` (new) | 4h | | 14 | **Build status page CRUD API** (create/edit/delete pages) | `app/api/pages.py` (new) | 3h | | 15 | **Write multi-tenancy tests** (data isolation, tenant-scoped CRUD) | `tests/test_api_tenancy.py` (new) | 3h | ### Phase 3: Feature Tiers (Week 3–4) | # | Step | Files | Effort | |---|---|---|---| | 16 | **Create `tier_limits.py`** with all limit definitions and enforcement | `app/services/tier_limits.py` (new) | 2h | | 17 | **Integrate limit checks** into all write endpoints | `app/api/services.py`, `monitors.py`, `subscribers.py`, `pages.py`, `organizations.py` | 3h | | 18 | **Build billing UI** — plan selection page with upgrade CTAs | `app/templates/billing.html` (new), `app/static/css/style.css` | 3h | | 19 | **Add billing route** that serves the billing page with org context | `app/main.py` (add route) | 1h | | 20 | **Write tier limit tests** (enforce limits, verify free/pro/team boundaries) | `tests/test_tier_limits.py` (new) | 3h | ### Phase 4: Stripe Payments (Week 4–5) | # | Step | Files | Effort | |---|---|---|---| | 21 | **Create Stripe Payment Links** in Stripe Dashboard (manual, no code) | Stripe Dashboard | 0.5h | | 22 | **Add Stripe config** to settings | `app/config.py` | 0.5h | | 23 | **Build billing API** — checkout redirect + webhook handler | `app/api/billing.py` (new), `app/api/router.py` | 4h | | 24 | **Add stripe Python SDK** to dependencies | `pyproject.toml` (add `stripe>=8.0`) | 0.5h | | 25 | **Configure Stripe webhook endpoint** in Stripe Dashboard → `https://korpo.pro/api/v1/billing/webhook` | Stripe Dashboard | 0.5h | | 26 | **Build success/cancel pages** | `app/templates/billing_success.html` (new) | 1h | | 27 | **Write webhook tests** with mock Stripe events | `tests/test_api_billing.py` (new) | 3h | ### Phase 5: Custom Domains & Polish (Week 5–6) | # | Step | Files | Effort | |---|---|---|---| | 28 | **Add custom domain middleware** | `app/middleware.py` (new), `app/main.py` (register) | 3h | | 29 | **Build landing page** at `GET /` (replace current global status route) | `app/templates/landing.html` (new), `app/main.py` | 4h | | 30 | **Add admin dashboard** — overview of org's pages, services, incidents | `app/templates/dashboard.html` (new), `app/api/dashboard.py` (new) | 6h | | 31 | **Proper PostgreSQL migration** (migration 004) | `pyproject.toml`, `app/config.py`, `docker-compose.yml` (add postgres service) | 4h | | 32 | **Nginx update** — wildcard subdomain or custom domain routing | nginx config | 2h | | 33 | **E2E test suite** — full user journey from signup → billing → status page | `tests/test_e2e.py` (new) | 4h | ### Phase 6: Launch (Week 6–7) | # | Step | Files | Effort | |---|---|---|---| | 34 | **Add `python-jose` + `passlib[bcrypt]` + `stripe`** to deps | `pyproject.toml` | 0.5h | | 35 | **Rate limiting middleware** for public endpoints | `app/middleware.py` | 2h | | 36 | **Email verification flow** for user signups | `app/api/auth.py`, `app/services/notifier.py` | 3h | | 37 | **CORS configuration** for API access (Pro+ feature) | `app/main.py` | 1h | | 38 | **Production deployment** — update systemd service, env vars, DB | `indie-status-page.service` (env file), nginx | 2h | | 39 | **Documentation** — API docs, self-serve guide | `README.md` update, `docs/` | 4h | --- ## Appendix A: New Dependencies Add to `pyproject.toml`: ```toml dependencies = [ # ... existing ... "python-jose[cryptography]>=3.3,<4.0", # JWT tokens "passlib[bcrypt]>=1.7,<2.0", # Password hashing "python-multipart>=0.0.6,<1.0", # OAuth2 form parsing (already present) "stripe>=8.0,<9.0", # Stripe SDK (webhook verification) "email-validator>=2.1,<3.0", # Pydantic EmailStr validation ] # For PostgreSQL (Phase 5): # "asyncpg>=0.29,<1.0", ``` ## Appendix B: Updated `app/models/__init__.py` ```python from app.models.models import ( User, Organization, OrganizationMember, StatusPage, Service, Incident, IncidentUpdate, Monitor, MonitorResult, Subscriber, NotificationLog, SiteSetting, ) __all__ = [ "User", "Organization", "OrganizationMember", "StatusPage", "Service", "Incident", "IncidentUpdate", "Monitor", "MonitorResult", "Subscriber", "NotificationLog", "SiteSetting", ] ``` ## Appendix C: Router Registration Updates ```python # app/api/router.py — updated from fastapi import APIRouter from app.api.auth import router as auth_router from app.api.billing import router as billing_router from app.api.services import router as services_router from app.api.incidents import router as incidents_router from app.api.monitors import router as monitors_router from app.api.subscribers import router as subscribers_router from app.api.settings import router as settings_router from app.api.pages import router as pages_router # NEW from app.api.organizations import router as org_router # NEW from app.api.dashboard import router as dashboard_router # NEW api_v1_router = APIRouter() api_v1_router.include_router(auth_router, prefix="/auth", tags=["auth"]) api_v1_router.include_router(billing_router, prefix="/billing", tags=["billing"]) api_v1_router.include_router(services_router, prefix="/services", tags=["services"]) api_v1_router.include_router(incidents_router, prefix="/incidents", tags=["incidents"]) api_v1_router.include_router(monitors_router, prefix="/monitors", tags=["monitors"]) api_v1_router.include_router(subscribers_router, prefix="/subscribers", tags=["subscribers"]) api_v1_router.include_router(settings_router, prefix="/settings", tags=["settings"]) api_v1_router.include_router(pages_router, prefix="/pages", tags=["pages"]) api_v1_router.include_router(org_router, prefix="/organizations", tags=["organizations"]) api_v1_router.include_router(dashboard_router, prefix="/dashboard", tags=["dashboard"]) ``` ## Appendix D: Key File Map (New + Modified) | File | Action | Phase | |---|---|---| | `app/models/models.py` | **Modify** — add User, Organization, OrganizationMember, StatusPage; add organization_id to existing models | 1–2 | | `app/dependencies.py` | **Modify** — add get_current_user, get_current_org | 1 | | `app/config.py` | **Modify** — add Stripe URLs, JWT secret fields | 1, 4 | | `app/api/auth.py` | **Create** — register, login, JWT logic | 1 | | `app/api/billing.py` | **Create** — checkout redirect, webhook, success | 4 | | `app/api/pages.py` | **Create** — StatusPage CRUD | 2 | | `app/api/organizations.py` | **Create** — org management, invite | 2 | | `app/api/dashboard.py` | **Create** — admin dashboard API | 5 | | `app/services/tier_limits.py` | **Create** — tier enforcement | 3 | | `app/middleware.py` | **Create** — custom domain resolution | 5 | | `app/api/services.py` | **Modify** — add org scoping + tier limit checks | 2 | | `app/api/incidents.py` | **Modify** — add org scoping | 2 | | `app/api/monitors.py` | **Modify** — add org scoping + tier limit checks | 2 | | `app/api/subscribers.py` | **Modify** — add org scoping + tier limit checks | 2 | | `app/api/settings.py` | **Modify** — add org scoping | 2 | | `app/api/router.py` | **Modify** — register new routers | 1, 2, 4 | | `app/main.py` | **Modify** — add multi-page routes, landing, middleware, dashboard | 2, 5 | | `app/templates/billing.html` | **Create** — plan selection page | 3 | | `app/templates/landing.html` | **Create** — marketing homepage | 5 | | `app/templates/dashboard.html` | **Create** — admin dashboard | 5 | | `app/templates/billing_success.html` | **Create** — post-payment success | 4 | | `app/static/css/style.css` | **Modify** — add billing, dashboard, plan card styles | 3 | | `pyproject.toml` | **Modify** — add new dependencies | 1, 4 | | `alembic.ini` | **Modify** — proper config | 1 | | `migrations/env.py` | **Modify** — import Base + models | 1 | | `migrations/versions/001_*.py` | **Create** — user/org tables | 1 | | `migrations/versions/002_*.py` | **Create** — organization_id on existing tables | 2 | | `migrations/versions/003_*.py` | **Create** — data backfill | 2 | | `docker-compose.yml` | **Modify** — add PostgreSQL service | 5 | | `tests/test_api_auth.py` | **Create** | 1 | | `tests/test_api_tenancy.py` | **Create** | 2 | | `tests/test_tier_limits.py` | **Create** | 3 | | `tests/test_api_billing.py` | **Create** | 4 | | `tests/test_e2e.py` | **Create** | 5 | --- *Plan generated: 2026-04-25 | Total estimated effort: ~6–7 weeks for a solo developer*