Add invite code registration system for beta

Backend:
- Add InviteCode model with single-use codes
- Add invite API endpoints (create, list, revoke, validate)
- Modify registration to require invite code when enabled
- Add REQUIRE_INVITE_CODE config toggle (default: true)
- Add Alembic migration for invite_codes table

Frontend:
- Add invite code field to registration page
- Validate invite code on blur with visual feedback
- Pass invite code to registration API

Admins can generate invite codes via /api/docs (Swagger UI).

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Michael Chihlas
2026-02-01 00:08:06 -05:00
parent 005db0700c
commit 20c4c40a1f
16 changed files with 412 additions and 4 deletions

View File

@@ -5,6 +5,7 @@ from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.config import settings
from app.core.database import get_db
from app.core.security import (
verify_password,
@@ -14,6 +15,7 @@ from app.core.security import (
decode_token
)
from app.models.user import User
from app.models.invite_code import InviteCode
from app.schemas.user import UserCreate, UserResponse, UserLogin
from app.schemas.token import Token
from app.api.deps import get_current_user
@@ -27,6 +29,39 @@ async def register(
db: Annotated[AsyncSession, Depends(get_db)]
):
"""Register a new user."""
# Validate invite code if required
invite_code_record = None
if settings.REQUIRE_INVITE_CODE:
if not user_data.invite_code:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invite code is required"
)
# Look up invite code (case-insensitive)
result = await db.execute(
select(InviteCode).where(InviteCode.code == user_data.invite_code.upper())
)
invite_code_record = result.scalar_one_or_none()
if not invite_code_record:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid invite code"
)
if invite_code_record.is_used:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invite code has already been used"
)
if invite_code_record.is_expired:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invite code has expired"
)
# Check if email already exists
result = await db.execute(select(User).where(User.email == user_data.email))
existing_user = result.scalar_one_or_none()
@@ -41,9 +76,16 @@ async def register(
email=user_data.email,
password_hash=get_password_hash(user_data.password),
name=user_data.name,
role=user_data.role # Use role from request (defaults to "engineer")
role=user_data.role,
invite_code_id=invite_code_record.id if invite_code_record else None
)
db.add(new_user)
# Mark invite code as used
if invite_code_record:
invite_code_record.used_by_id = new_user.id
invite_code_record.used_at = datetime.now(timezone.utc)
await db.commit()
await db.refresh(new_user)

View File

@@ -0,0 +1,96 @@
from datetime import datetime, timezone
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.models.user import User
from app.models.invite_code import InviteCode
from app.schemas.invite_code import InviteCodeCreate, InviteCodeResponse, InviteCodeValidation
from app.api.deps import require_admin
router = APIRouter(prefix="/invites", tags=["invites"])
@router.post("", response_model=InviteCodeResponse, status_code=status.HTTP_201_CREATED)
async def create_invite_code(
invite_data: InviteCodeCreate,
current_user: Annotated[User, Depends(require_admin)],
db: Annotated[AsyncSession, Depends(get_db)]
):
"""Create a new invite code. Admin only."""
invite_code = InviteCode(
created_by_id=current_user.id,
expires_at=invite_data.expires_at,
note=invite_data.note
)
db.add(invite_code)
await db.commit()
await db.refresh(invite_code)
return invite_code
@router.get("", response_model=list[InviteCodeResponse])
async def list_invite_codes(
current_user: Annotated[User, Depends(require_admin)],
db: Annotated[AsyncSession, Depends(get_db)]
):
"""List all invite codes. Admin only."""
result = await db.execute(
select(InviteCode).order_by(InviteCode.created_at.desc())
)
invite_codes = result.scalars().all()
return invite_codes
@router.delete("/{code}", status_code=status.HTTP_204_NO_CONTENT)
async def revoke_invite_code(
code: str,
current_user: Annotated[User, Depends(require_admin)],
db: Annotated[AsyncSession, Depends(get_db)]
):
"""Revoke (delete) an invite code. Admin only."""
result = await db.execute(
select(InviteCode).where(InviteCode.code == code)
)
invite_code = result.scalar_one_or_none()
if not invite_code:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Invite code not found"
)
if invite_code.is_used:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot revoke a used invite code"
)
await db.delete(invite_code)
await db.commit()
@router.get("/validate/{code}", response_model=InviteCodeValidation)
async def validate_invite_code(
code: str,
db: Annotated[AsyncSession, Depends(get_db)]
):
"""Check if an invite code is valid. Public endpoint for UX."""
result = await db.execute(
select(InviteCode).where(InviteCode.code == code.upper())
)
invite_code = result.scalar_one_or_none()
if not invite_code:
return InviteCodeValidation(valid=False, message="Invalid invite code")
if invite_code.is_used:
return InviteCodeValidation(valid=False, message="Invite code has already been used")
if invite_code.is_expired:
return InviteCodeValidation(valid=False, message="Invite code has expired")
return InviteCodeValidation(valid=True, message="Invite code is valid")