graph-requirements-planner/.planning/research/ARCHITECTURE.md

Architecture Research: Multi-Tenancy & User Accounts Integration

Domain: Requirements management system (FastAPI + SQLAlchemy 2.0 + SQLite) Researched: 2026-03-18 Confidence: HIGH

Executive Summary

Adding user accounts and multi-tenancy to an existing FastAPI + SQLAlchemy + SQLite application requires four architectural layers: Authentication (user identity), Authorization (permissions), Tenant Isolation (data separation), and Migration Strategy (backwards compatibility). The recommended approach uses shared-schema multi-tenancy with tenant-scoped queries, JWT tokens with refresh tokens, RBAC via junction tables, and FastAPI's dependency injection for context management.

Key decision: Use shared-schema multi-tenancy (single database, user_id + tenant_id filtering) rather than database-per-tenant, because SQLite's single-writer lock makes multiple active databases perform worse than filtered queries on a single database.

Existing Architecture Analysis

Current State

┌─────────────────────────────────────────────────────────────┐
│                     Angular 17 Frontend                      │
│  - AuthGuard + HTTP Interceptor for token injection         │
│  - localStorage for token storage                           │
└─────────────────────┬───────────────────────────────────────┘
                      │ HTTP/JSON
┌─────────────────────┴───────────────────────────────────────┐
│                      FastAPI Backend                         │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  auth.py: Single password → JWT (24h, no refresh)     │  │
│  │  verify_token() dependency on all protected routes    │  │
│  └────────────────────────────────────────────────────────┘  │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Routers: requirements, projects, links, layouts,     │  │
│  │           stats (all protected by verify_token)       │  │
│  └────────────────────────────────────────────────────────┘  │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  SQLAlchemy 2.0 ORM with get_db() dependency          │  │
│  └────────────────────────────────────────────────────────┘  │
└─────────────────────┬───────────────────────────────────────┘
                      │
┌─────────────────────┴───────────────────────────────────────┐
│                    SQLite Database                           │
│  ┌──────────────────┐  ┌──────────────────┐                 │
│  │ requirement_nodes│  │ projects         │                 │
│  │ (no user_id)     │  │ (no user_id)     │                 │
│  └──────────────────┘  └──────────────────┘                 │
│  ┌──────────────────┐  ┌──────────────────┐                 │
│  │ cross_pillar_links│  │ requirement_history│              │
│  │                  │  │ (no user_id)     │                 │
│  └──────────────────┘  └──────────────────┘                 │
│  ┌──────────────────┐                                        │
│  │ graph_layouts    │                                        │
│  │ (no user_id)     │                                        │
│  └──────────────────┘                                        │
└─────────────────────────────────────────────────────────────┘

Architecture Gaps

Component	Current State	Required for v2.0
Authentication	Single password, 24h JWT	User accounts with email/password, refresh tokens
User Model	None	Users table with hashed passwords
Tenant Isolation	None (all data shared)	Per-user project ownership, user_id filtering
Authorization	Binary (authenticated or not)	Role-based access (owner/editor/viewer per project)
Data Attribution	No edit tracking	User attribution in history
Invite System	N/A	Invite-only registration with tokens
Session Management	Stateless JWT (24h only)	Refresh tokens (long-lived, revocable)

Recommended Architecture

Target State with Multi-Tenancy

┌─────────────────────────────────────────────────────────────┐
│                     Angular 17 Frontend                      │
│  - AuthGuard with user profile                              │
│  - HTTP Interceptor with token refresh logic                │
│  - localStorage: access_token + refresh_token               │
└─────────────────────┬───────────────────────────────────────┘
                      │ HTTP/JSON
┌─────────────────────┴───────────────────────────────────────┐
│                      FastAPI Backend                         │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Authentication Layer (auth.py + new users.py)         │  │
│  │  - POST /auth/register (invite token required)         │  │
│  │  - POST /auth/login (email + password → tokens)        │  │
│  │  - POST /auth/refresh (refresh_token → new access)     │  │
│  │  - POST /auth/logout (revoke refresh_token)            │  │
│  │  - GET /auth/me (current user profile)                 │  │
│  └────────────────────────────────────────────────────────┘  │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Dependency Injection Context                           │  │
│  │  get_current_user() → UserContext (user_id, email)     │  │
│  │  require_project_access(role) → enforces RBAC          │  │
│  └────────────────────────────────────────────────────────┘  │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Tenant-Scoped Query Middleware                         │  │
│  │  Auto-inject user_id filter on all queries             │  │
│  │  (via SQLAlchemy session events)                        │  │
│  └────────────────────────────────────────────────────────┘  │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Routers (modified for user context)                    │  │
│  │  - requirements: add created_by, updated_by            │  │
│  │  - projects: add owner_id, check permissions           │  │
│  │  - links, layouts, stats: scope to accessible projects │  │
│  └────────────────────────────────────────────────────────┘  │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  SQLAlchemy 2.0 ORM                                     │  │
│  │  get_db_with_tenant_filter() → scoped session          │  │
│  └────────────────────────────────────────────────────────┘  │
└─────────────────────┬───────────────────────────────────────┘
                      │
┌─────────────────────┴───────────────────────────────────────┐
│                    SQLite Database                           │
│  ┌──────────────────┐  ┌──────────────────┐                 │
│  │ users            │  │ projects         │                 │
│  │ (NEW)            │  │ + owner_id (FK)  │                 │
│  └──────────────────┘  └──────────────────┘                 │
│  ┌──────────────────┐  ┌──────────────────┐                 │
│  │ project_members  │  │ requirement_nodes│                 │
│  │ (NEW - RBAC)     │  │ + created_by_id  │                 │
│  │                  │  │ + updated_by_id  │                 │
│  └──────────────────┘  └──────────────────┘                 │
│  ┌──────────────────┐  ┌──────────────────┐                 │
│  │ invite_tokens    │  │ requirement_history│              │
│  │ (NEW)            │  │ + user_id        │                 │
│  └──────────────────┘  └──────────────────┘                 │
│  ┌──────────────────┐  ┌──────────────────┐                 │
│  │ refresh_tokens   │  │ graph_layouts    │                 │
│  │ (NEW)            │  │ (no change)      │                 │
│  └──────────────────┘  └──────────────────┘                 │
│  ┌──────────────────┐                                        │
│  │ cross_pillar_links│ (no change - inherits access from   │
│  │                  │  parent requirement)                  │
│  └──────────────────┘                                        │
└─────────────────────────────────────────────────────────────┘

New Database Schema

New Tables

1. users (Core Identity)

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    full_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    is_admin: Mapped[bool] = mapped_column(Boolean, default=False)  # Global admin flag
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
    last_login: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    # Relationships
    owned_projects: Mapped[list["Project"]] = relationship("Project", back_populates="owner")
    project_memberships: Mapped[list["ProjectMember"]] = relationship(
        "ProjectMember", back_populates="user", cascade="all, delete-orphan"
    )
    created_requirements: Mapped[list["RequirementNode"]] = relationship(
        "RequirementNode", foreign_keys="RequirementNode.created_by_id", back_populates="created_by"
    )
    refresh_tokens: Mapped[list["RefreshToken"]] = relationship(
        "RefreshToken", back_populates="user", cascade="all, delete-orphan"
    )

__table_args__ = (
    Index("idx_users_email_active", "email", "is_active"),  # Fast login lookup
)

Why this design:

• email as primary identity (unique, indexed for fast login)
• is_admin for global operations (e.g., invite token generation)
• is_active for soft account suspension without data loss
• last_login for monitoring and security auditing

2. project_members (RBAC Junction Table)

class ProjectMemberRole(str, enum.Enum):
    owner = "owner"    # Full control, can delete project
    editor = "editor"  # Can modify requirements, cannot manage members
    viewer = "viewer"  # Read-only access

class ProjectMember(Base):
    __tablename__ = "project_members"

    project_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("projects.id", ondelete="CASCADE"), primary_key=True
    )
    user_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
    )
    role: Mapped[ProjectMemberRole] = mapped_column(
        Enum(ProjectMemberRole), nullable=False, default=ProjectMemberRole.viewer
    )
    added_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
    added_by_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True
    )

    # Relationships
    project: Mapped["Project"] = relationship("Project", back_populates="members")
    user: Mapped["User"] = relationship("User", back_populates="project_memberships")

__table_args__ = (
    Index("idx_project_members_user", "user_id"),  # Find all projects for a user
    Index("idx_project_members_project", "project_id"),  # Find all members of a project
)

Why this design:

• Composite primary key (project_id, user_id) prevents duplicate memberships
• Three-tier role model (owner/editor/viewer) covers most collaboration patterns
• added_by_id for audit trail (who invited whom)
• Cascade delete ensures cleanup when projects or users are deleted

3. invite_tokens (Invite-Only Registration)

class InviteToken(Base):
    __tablename__ = "invite_tokens"

    token: Mapped[str] = mapped_column(String(64), primary_key=True)  # UUID or secure random
    email: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
    created_by_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False
    )
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    used_by_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True
    )

__table_args__ = (
    Index("idx_invite_tokens_email_used", "email", "used_at"),  # Check if email already invited
    Index("idx_invite_tokens_expires", "expires_at"),  # Cleanup expired tokens
)

Why this design:

• Token as primary key (opaque, unpredictable)
• email scoped invites prevent multiple registrations with same token
• expires_at for time-limited invites (e.g., 7 days)
• used_at / used_by_id track invite consumption (one-time use)
• Created by existing users (only authenticated users can invite)

4. refresh_tokens (Session Management)

class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
    )
    token_hash: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    # Relationships
    user: Mapped["User"] = relationship("User", back_populates="refresh_tokens")

__table_args__ = (
    Index("idx_refresh_tokens_user_active", "user_id", "revoked_at"),  # Active tokens per user
    Index("idx_refresh_tokens_expires", "expires_at"),  # Cleanup expired tokens
)

Why this design:

• token_hash stores hashed refresh token (never store plaintext tokens in DB)
• expires_at for long-lived but not permanent tokens (e.g., 30 days)
• revoked_at enables explicit logout (revoke token)
• last_used_at for security monitoring (detect stolen tokens via unusual usage patterns)
• Multiple refresh tokens per user allowed (multi-device support)

Modified Existing Tables

projects (Add Ownership)

# ADD to existing Project model:
owner_id: Mapped[int] = mapped_column(
    Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)

owner: Mapped["User"] = relationship("User", back_populates="owned_projects")
members: Mapped[list["ProjectMember"]] = relationship(
    "ProjectMember", back_populates="project", cascade="all, delete-orphan"
)

Migration strategy:

• New column owner_id with server_default=1 (assumes first admin user has id=1)
• Run data migration to assign existing projects to the admin user
• After migration, make owner_id NOT NULL

requirement_nodes (Add Attribution)

# ADD to existing RequirementNode model:
created_by_id: Mapped[int] = mapped_column(
    Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)
updated_by_id: Mapped[int | None] = mapped_column(
    Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)

created_by: Mapped["User | None"] = relationship(
    "User", foreign_keys=[created_by_id], back_populates="created_requirements"
)
updated_by: Mapped["User | None"] = relationship(
    "User", foreign_keys=[updated_by_id]
)

Migration strategy:

• Both columns nullable (existing data has NULL creator)
• SET NULL on user deletion preserves history even if user deleted
• Display "Unknown User" in UI when created_by_id is NULL

requirement_history (Add User Attribution)

# ADD to existing RequirementHistory model:
user_id: Mapped[int | None] = mapped_column(
    Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)

user: Mapped["User | None"] = relationship("User")

Migration strategy:

• Nullable because historical edits have no user attribution
• New history entries always have user_id (enforced in router logic)

Integration Patterns

Pattern 1: User Context Injection (Dependency Injection)

What: Extract authenticated user from JWT and provide as dependency to route handlers.

Implementation:

# In auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from pydantic import BaseModel

security = HTTPBearer()

class UserContext(BaseModel):
    user_id: int
    email: str
    is_admin: bool

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db)
) -> UserContext:
    """Extract user from JWT access token."""
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid or expired token")

    # Verify user exists and is active
    user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
    if not user:
        raise HTTPException(status_code=401, detail="User not found or inactive")

    return UserContext(user_id=user.id, email=user.email, is_admin=user.is_admin)

# Usage in routes:
@router.get("/api/projects")
def list_projects(
    db: Session = Depends(get_db),
    current_user: UserContext = Depends(get_current_user)
):
    # current_user available here
    pass

Why this works:

• FastAPI's dependency injection runs get_current_user before route handler
• Raises HTTPException if token invalid (automatic 401 response)
• UserContext is immutable (Pydantic model) preventing tampering
• No middleware needed—explicit dependencies are clearer

Pattern 2: RBAC Authorization (Permission Checking)

What: Check user's role on a project before allowing operations.

Implementation:

# In auth.py
from functools import wraps
from typing import Callable

class PermissionDenied(HTTPException):
    def __init__(self):
        super().__init__(status_code=403, detail="Insufficient permissions")

def require_project_access(
    project_id: int,
    required_role: ProjectMemberRole,
    db: Session,
    current_user: UserContext
) -> ProjectMember:
    """
    Check if user has required role on project.
    Returns ProjectMember if authorized, raises 403 otherwise.

    Role hierarchy: owner > editor > viewer
    """
    # Global admins bypass all checks
    if current_user.is_admin:
        return ProjectMember(
            project_id=project_id,
            user_id=current_user.user_id,
            role=ProjectMemberRole.owner
        )

    # Check ownership
    project = db.query(Project).filter(Project.id == project_id).first()
    if not project:
        raise HTTPException(status_code=404, detail="Project not found")

    if project.owner_id == current_user.user_id:
        return ProjectMember(
            project_id=project_id,
            user_id=current_user.user_id,
            role=ProjectMemberRole.owner
        )

    # Check membership
    member = db.query(ProjectMember).filter(
        ProjectMember.project_id == project_id,
        ProjectMember.user_id == current_user.user_id
    ).first()

    if not member:
        raise PermissionDenied()

    # Role hierarchy check
    role_order = {
        ProjectMemberRole.viewer: 1,
        ProjectMemberRole.editor: 2,
        ProjectMemberRole.owner: 3,
    }

    if role_order[member.role] < role_order[required_role]:
        raise PermissionDenied()

    return member

# Usage in routes:
@router.delete("/api/requirements/{req_id}")
def delete_requirement(
    req_id: str,
    db: Session = Depends(get_db),
    current_user: UserContext = Depends(get_current_user)
):
    req = db.query(RequirementNode).filter(RequirementNode.id == req_id).first()
    if not req:
        raise HTTPException(status_code=404)

    # Check editor access on project
    require_project_access(
        project_id=req.project_id,
        required_role=ProjectMemberRole.editor,
        db=db,
        current_user=current_user
    )

    # Authorized—proceed with deletion
    db.delete(req)
    db.commit()

Why this works:

• Explicit permission checks at operation level (not blanket middleware)
• Role hierarchy encoded in single function (DRY principle)
• Global admins bypass for administrative operations
• Project owners inherit full access automatically

Pattern 3: Tenant-Scoped Queries (Data Isolation)

What: Automatically filter queries to only return data user has access to.

Implementation (Option A: Manual Filtering - Recommended for SQLite):

# In routers/projects.py
@router.get("/api/projects", response_model=list[ProjectResponse])
def list_projects(
    db: Session = Depends(get_db),
    current_user: UserContext = Depends(get_current_user)
):
    """List all projects accessible to current user."""
    if current_user.is_admin:
        # Admins see all projects
        projects = db.query(Project).all()
    else:
        # Users see owned + member projects
        owned = db.query(Project).filter(Project.owner_id == current_user.user_id)
        member = (
            db.query(Project)
            .join(ProjectMember)
            .filter(ProjectMember.user_id == current_user.user_id)
        )
        projects = owned.union(member).all()

    return projects

Implementation (Option B: SQLAlchemy Session Event Hooks - Advanced):

# In database.py
from sqlalchemy import event
from sqlalchemy.orm import Session

class TenantSession(Session):
    """Session subclass that auto-filters queries by user_id."""
    def __init__(self, user_id: int | None = None, **kwargs):
        super().__init__(**kwargs)
        self.user_id = user_id

@event.listens_for(TenantSession, "do_orm_execute")
def receive_do_orm_execute(execute_state):
    """Intercept queries and inject tenant filters."""
    if execute_state.is_select and execute_state.session.user_id:
        # Auto-add user_id filter to all queries on tables with user_id column
        # (Complex implementation—see SQLAlchemy docs)
        pass

# In dependency:
def get_tenant_db(current_user: UserContext = Depends(get_current_user)):
    db = TenantSession(user_id=current_user.user_id, bind=engine)
    try:
        yield db
    finally:
        db.close()

Recommendation: Use Option A (Manual Filtering) for this project because:

• SQLite doesn't support Row Level Security like PostgreSQL
• Explicit filters are easier to understand and debug
• Session event hooks add complexity and can break with ORM updates
• Project has moderate query count—manual filters are manageable

Pattern 4: Refresh Token Rotation (Session Security)

What: Short-lived access tokens (15min) with long-lived refresh tokens (30 days).

Implementation:

# In auth.py
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_DAYS = 30

def create_access_token(user_id: int) -> str:
    """Short-lived token for API requests."""
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": user_id, "exp": expire, "type": "access"}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(db: Session, user_id: int) -> str:
    """Long-lived token for obtaining new access tokens."""
    token = secrets.token_urlsafe(32)  # Cryptographically secure random
    token_hash = hashlib.sha256(token.encode()).hexdigest()

    expires_at = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)

    db_token = RefreshToken(
        user_id=user_id,
        token_hash=token_hash,
        expires_at=expires_at
    )
    db.add(db_token)
    db.commit()

    return token  # Return plaintext (only time it's visible)

@router.post("/auth/refresh", response_model=TokenResponse)
def refresh_access_token(
    refresh_token: str,
    db: Session = Depends(get_db)
):
    """Exchange refresh token for new access token."""
    token_hash = hashlib.sha256(refresh_token.encode()).hexdigest()

    db_token = db.query(RefreshToken).filter(
        RefreshToken.token_hash == token_hash,
        RefreshToken.revoked_at.is_(None),
        RefreshToken.expires_at > datetime.now(timezone.utc)
    ).first()

    if not db_token:
        raise HTTPException(status_code=401, detail="Invalid or expired refresh token")

    # Update last used timestamp
    db_token.last_used_at = datetime.now(timezone.utc)
    db.commit()

    # Generate new access token
    access_token = create_access_token(db_token.user_id)

    return TokenResponse(access_token=access_token)

@router.post("/auth/logout")
def logout(
    refresh_token: str,
    db: Session = Depends(get_db),
    current_user: UserContext = Depends(get_current_user)
):
    """Revoke refresh token (logout)."""
    token_hash = hashlib.sha256(refresh_token.encode()).hexdigest()

    db_token = db.query(RefreshToken).filter(
        RefreshToken.token_hash == token_hash,
        RefreshToken.user_id == current_user.user_id
    ).first()

    if db_token:
        db_token.revoked_at = datetime.now(timezone.utc)
        db.commit()

    return {"message": "Logged out"}

Frontend integration (Angular):

// In auth.interceptor.ts
intercept(req: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
  return next.handle(this.addToken(req)).pipe(
    catchError((error: HttpErrorResponse) => {
      if (error.status === 401 && this.getRefreshToken()) {
        return this.handle401Error(req, next);
      }
      return throwError(() => error);
    })
  );
}

private handle401Error(req: HttpRequest<any>, next: HttpHandler) {
  if (!this.isRefreshing) {
    this.isRefreshing = true;
    this.refreshTokenSubject.next(null);

    return this.authService.refreshToken().pipe(
      switchMap((token: any) => {
        this.isRefreshing = false;
        this.refreshTokenSubject.next(token.access_token);
        return next.handle(this.addToken(req));
      }),
      catchError((err) => {
        this.isRefreshing = false;
        this.authService.logout();
        return throwError(() => err);
      })
    );
  }

  return this.refreshTokenSubject.pipe(
    filter(token => token !== null),
    take(1),
    switchMap(() => next.handle(this.addToken(req)))
  );
}

Why this works:

• Access tokens short-lived (minimal damage if stolen)
• Refresh tokens stored in httpOnly cookies or secure storage (harder to steal)
• Token rotation prevents replay attacks
• Explicit revocation enables secure logout

Pattern 5: Invite-Only Registration Flow

What: Only users with valid invite tokens can register.

Implementation:

# In auth.py
@router.post("/auth/invites", dependencies=[Depends(get_current_user)])
def create_invite(
    email: str,
    db: Session = Depends(get_db),
    current_user: UserContext = Depends(get_current_user)
):
    """Create invite token (authenticated users only)."""
    # Check if email already registered
    if db.query(User).filter(User.email == email).first():
        raise HTTPException(status_code=400, detail="User already exists")

    # Check for unused invite
    existing = db.query(InviteToken).filter(
        InviteToken.email == email,
        InviteToken.used_at.is_(None),
        InviteToken.expires_at > datetime.now(timezone.utc)
    ).first()

    if existing:
        raise HTTPException(status_code=400, detail="Invite already sent")

    # Generate token
    token = secrets.token_urlsafe(32)
    expires_at = datetime.now(timezone.utc) + timedelta(days=7)

    invite = InviteToken(
        token=token,
        email=email,
        created_by_id=current_user.user_id,
        expires_at=expires_at
    )
    db.add(invite)
    db.commit()

    # TODO: Send email with registration link
    # send_invite_email(email, f"https://app.example.com/register?token={token}")

    return {"token": token, "expires_at": expires_at}

@router.post("/auth/register", response_model=TokenResponse)
def register(
    invite_token: str,
    email: str,
    password: str,
    full_name: str | None,
    db: Session = Depends(get_db)
):
    """Register new user with invite token."""
    # Validate invite
    invite = db.query(InviteToken).filter(
        InviteToken.token == invite_token,
        InviteToken.email == email,
        InviteToken.used_at.is_(None),
        InviteToken.expires_at > datetime.now(timezone.utc)
    ).first()

    if not invite:
        raise HTTPException(status_code=400, detail="Invalid or expired invite")

    # Check if user exists
    if db.query(User).filter(User.email == email).first():
        raise HTTPException(status_code=400, detail="User already exists")

    # Create user
    hashed_password = get_password_hash(password)
    user = User(
        email=email,
        hashed_password=hashed_password,
        full_name=full_name
    )
    db.add(user)

    # Mark invite as used
    invite.used_at = datetime.now(timezone.utc)
    invite.used_by_id = user.id

    db.commit()

    # Generate tokens
    access_token = create_access_token(user.id)
    refresh_token = create_refresh_token(db, user.id)

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_token
    )

Why this works:

• Email scoped invites prevent token reuse
• Expiration limits token lifespan
• Audit trail tracks who invited whom
• One-time use prevents invite sharing

Data Flow Changes

Before (v1.0 - Single Password)

User enters password
    ↓
POST /api/auth/login {password}
    ↓
Backend validates against single PASSWORD env var
    ↓
If valid: JWT created with {exp} only
    ↓
Frontend stores token in localStorage
    ↓
All API requests include: Authorization: Bearer <token>
    ↓
verify_token() checks JWT signature and expiration
    ↓
If valid: route handler executes (no user context)

After (v2.0 - User Accounts)

User enters email + password
    ↓
POST /api/auth/login {email, password}
    ↓
Backend queries users table by email
    ↓
Verify hashed_password with bcrypt
    ↓
If valid: Generate access_token + refresh_token
    ↓
Frontend stores both tokens (localStorage or httpOnly cookie)
    ↓
API requests include: Authorization: Bearer <access_token>
    ↓
get_current_user() extracts user_id from JWT
    ↓
Queries users table to verify user still active
    ↓
Returns UserContext(user_id, email, is_admin)
    ↓
Route handler receives current_user dependency
    ↓
Permission checks against project_members table
    ↓
Queries auto-filtered by user's accessible projects
    ↓
Write operations tag created_by_id / updated_by_id

New: Token Refresh Flow

Access token expires (15min)
    ↓
Frontend receives 401 on API request
    ↓
Interceptor catches 401, calls POST /api/auth/refresh {refresh_token}
    ↓
Backend validates refresh_token (hash lookup in DB)
    ↓
Checks: not revoked, not expired, user active
    ↓
If valid: Generate new access_token (refresh_token unchanged)
    ↓
Updates last_used_at timestamp on refresh_token
    ↓
Returns new access_token
    ↓
Frontend retries original request with new token
    ↓
(Transparent to user—no re-login needed)

Migration Strategy (Backwards Compatibility)

Phase 1: Add New Tables (Non-Breaking)

# Alembic migration: add users, project_members, invite_tokens, refresh_tokens
# No existing data affected—tables are empty initially

Phase 2: Add Columns with Nullable (Non-Breaking)

# Alembic migration:
# - ALTER TABLE projects ADD COLUMN owner_id INTEGER REFERENCES users(id)
# - ALTER TABLE requirement_nodes ADD COLUMN created_by_id INTEGER REFERENCES users(id)
# - ALTER TABLE requirement_nodes ADD COLUMN updated_by_id INTEGER REFERENCES users(id)
# - ALTER TABLE requirement_history ADD COLUMN user_id INTEGER REFERENCES users(id)

# All columns nullable initially—existing data has NULL

Phase 3: Seed Default Admin User

# Data migration script:
default_user = User(
    id=1,
    email="admin@localhost",
    hashed_password=get_password_hash(os.environ.get("ADMIN_PASSWORD", "changeme")),
    full_name="System Admin",
    is_admin=True
)
db.add(default_user)

# Assign all existing projects to admin
db.execute(
    update(Project).values(owner_id=1)
)

# Mark all existing requirements as created by admin
db.execute(
    update(RequirementNode)
    .where(RequirementNode.created_by_id.is_(None))
    .values(created_by_id=1)
)

db.commit()

Phase 4: Deploy New Auth Endpoints (Dual Auth)

# Keep old /api/auth/login {password} working
# Add new /api/auth/login {email, password}
# Frontend updated to use new endpoint
# Old endpoint deprecated but functional (gradual migration)

Phase 5: Make owner_id NOT NULL (Breaking)

# After confirming all projects have owner_id:
# Alembic migration:
# - ALTER TABLE projects ALTER COLUMN owner_id SET NOT NULL

Rollback Plan

If issues discovered after deployment:

1. Revert code: Old auth.py still validates tokens (SECRET_KEY unchanged)
2. Database intact: New columns nullable—old queries unaffected
3. No data loss: All existing data preserved with owner_id=1

Build Order (Recommended Sequence)

Milestone 1: Authentication Foundation

Goal: Users can register and login with email/password.

1. Create User model and table
2. Create InviteToken model and table
3. Create RefreshToken model and table
4. Implement password hashing (passlib + bcrypt)
5. Add /auth/register endpoint (with invite token validation)
6. Add /auth/login endpoint (email + password → tokens)
7. Add /auth/refresh endpoint (refresh token → new access token)
8. Add /auth/logout endpoint (revoke refresh token)
9. Update get_current_user() dependency
10. Seed default admin user in migration

Test: User can register with invite, login, and receive valid tokens.

Milestone 2: Project Ownership

Goal: Projects belong to users, users see only their projects.

1. Add owner_id column to projects (nullable)
2. Update project creation to set owner_id = current_user.user_id
3. Update GET /api/projects to filter by ownership
4. Update GET /api/requirements to filter by project ownership
5. Data migration: Assign existing projects to admin
6. Make owner_id NOT NULL

Test: User sees only projects they own; other users' projects hidden.

Milestone 3: RBAC (Project Sharing)

Goal: Project owners can invite collaborators with roles.

1. Create ProjectMember model and table
2. Add POST /api/projects/{id}/members (owner adds member)
3. Add GET /api/projects/{id}/members (list members)
4. Add DELETE /api/projects/{id}/members/{user_id} (remove member)
5. Implement require_project_access() dependency
6. Update all requirement routes to check RBAC
7. Add role field to project list response (user's role per project)

Test: Owner invites editor; editor can modify requirements but not delete project.

Milestone 4: Edit Attribution

Goal: History tracks who made each change.

1. Add created_by_id, updated_by_id to requirement_nodes
2. Add user_id to requirement_history
3. Update requirement creation to set created_by_id
4. Update requirement updates to set updated_by_id
5. Update history creation to set user_id
6. Update UI to display editor names in history

Test: History panel shows "Edited by Alice" instead of generic timestamps.

Milestone 5: Invite System

Goal: Users can invite others to join the platform.

1. Add POST /api/auth/invites (create invite token)
2. Add GET /api/auth/invites (list my sent invites)
3. Add invite email template
4. Integrate email sending (SMTP or service like Mailgun)
5. Update registration flow to send welcome email

Test: User creates invite; recipient registers successfully; both users see each other's projects if shared.

Anti-Patterns to Avoid

Anti-Pattern 1: Trusting Client-Provided user_id

What people do:

@router.post("/api/requirements")
def create_requirement(data: RequirementCreate, user_id: int):  # ❌ DANGEROUS
    node = RequirementNode(**data.dict(), created_by_id=user_id)
    # Attacker can impersonate anyone by sending any user_id

Why it's wrong: Clients can lie. Attacker sends user_id=1 to impersonate admin.

Do this instead:

@router.post("/api/requirements")
def create_requirement(
    data: RequirementCreate,
    current_user: UserContext = Depends(get_current_user)  # ✅ Server extracts from JWT
):
    node = RequirementNode(**data.dict(), created_by_id=current_user.user_id)

Anti-Pattern 2: Global Query Filters in Middleware

What people do:

# Middleware that silently filters all queries by user_id
@app.middleware("http")
async def tenant_filter_middleware(request, call_next):
    # Inject user_id filter into all SQLAlchemy queries
    # (Complex session hook magic)

Why it's wrong:

• Hidden behavior—hard to debug ("why is this query returning nothing?")
• Breaks when you need cross-tenant operations (e.g., admin views)
• SQLAlchemy session state leaks between requests
• Fails silently if middleware doesn't run (security vulnerability)

Do this instead:

# Explicit filtering in route handlers
@router.get("/api/projects")
def list_projects(current_user: UserContext = Depends(get_current_user), db: Session = Depends(get_db)):
    return db.query(Project).filter(
        (Project.owner_id == current_user.user_id) |
        (Project.members.any(ProjectMember.user_id == current_user.user_id))
    ).all()

Anti-Pattern 3: Storing Refresh Tokens in localStorage

What people do:

// Frontend stores refresh token in localStorage
localStorage.setItem('refresh_token', refreshToken);  // ❌ VULNERABLE TO XSS

Why it's wrong:

• XSS attacks can read localStorage and steal refresh tokens
• Refresh tokens are long-lived (30 days)—high-value targets

Do this instead:

// Store refresh token in httpOnly cookie (set by backend)
// Backend response:
Set-Cookie: refresh_token=<token>; HttpOnly; Secure; SameSite=Strict

// OR store in secure storage with encryption (mobile apps)
// Access tokens in memory only (SessionStorage acceptable)
sessionStorage.setItem('access_token', accessToken);  // ✅ Cleared on tab close

Anti-Pattern 4: No Token Expiration Validation

What people do:

def verify_token(token: str):
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    return payload["sub"]  # ❌ Doesn't check if user still exists/active

Why it's wrong:

• Deleted users can still authenticate (token valid until expiration)
• Deactivated accounts remain accessible
• No way to revoke access immediately

Do this instead:

def get_current_user(credentials = Depends(security), db: Session = Depends(get_db)):
    payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
    user_id = payload["sub"]

    user = db.query(User).filter(User.id == user_id, User.is_active == True).first()  # ✅ Verify user active
    if not user:
        raise HTTPException(status_code=401, detail="User not found or inactive")

    return UserContext(user_id=user.id, email=user.email, is_admin=user.is_admin)

Anti-Pattern 5: CASCADE DELETE Without Considering Foreign Keys

What people do:

# Delete user without considering owned projects
db.delete(user)  # ❌ Fails if projects reference this user
db.commit()

Why it's wrong:

• Foreign key constraint violations
• Data loss if CASCADE propagates unexpectedly
• No clear ownership transfer

Do this instead:

# Before deleting user, reassign or delete owned resources
def delete_user(user_id: int, db: Session):
    # Transfer project ownership to another admin
    admin = db.query(User).filter(User.is_admin == True, User.id != user_id).first()
    if not admin:
        raise HTTPException(status_code=400, detail="Cannot delete last admin")

    db.execute(
        update(Project)
        .where(Project.owner_id == user_id)
        .values(owner_id=admin.id)
    )

    # Now safe to delete user (SET NULL handles created_by/updated_by)
    user = db.query(User).filter(User.id == user_id).first()
    db.delete(user)
    db.commit()

Scaling Considerations

Single-Writer Limitation (SQLite)

Problem: SQLite uses a single write lock. With many concurrent users editing requirements, write contention increases.

Mitigation:

• Enable WAL mode: PRAGMA journal_mode=WAL; (allows concurrent reads during writes)
• Use connection pooling with conservative pool size (avoid lock queue buildup)
• Consider read replicas for queries (projects, stats dashboards)
• At ~50 concurrent writers, plan migration to PostgreSQL

When to migrate to PostgreSQL:

• User count: >100 active users (>50 concurrent writers)
• Data size: >10GB database
• Features needed: Full-text search, advanced RBAC (Row Level Security), JSON columns
• Cost: PostgreSQL adds operational complexity—delay until necessary

Token Cleanup (Garbage Collection)

Problem: Expired refresh tokens and invite tokens accumulate in database.

Solution:

# Scheduled task (run daily via cron or Celery)
def cleanup_expired_tokens(db: Session):
    now = datetime.now(timezone.utc)

    # Delete expired refresh tokens
    db.query(RefreshToken).filter(
        RefreshToken.expires_at < now
    ).delete()

    # Delete used or expired invites older than 30 days
    cutoff = now - timedelta(days=30)
    db.query(InviteToken).filter(
        (InviteToken.used_at < cutoff) | (InviteToken.expires_at < now)
    ).delete()

    db.commit()

Query Performance (Indexing)

Critical indexes added:

• users.email (unique, login queries)
• projects.owner_id (filter by owner)
• project_members(project_id, user_id) (composite PK, bidirectional lookups)
• requirement_nodes.project_id (filter requirements by project)
• refresh_tokens.token_hash (lookup during refresh)
• invite_tokens.email (check existing invites)

Monitor: Use EXPLAIN QUERY PLAN to verify index usage.

Security Considerations

Password Storage

• Hash algorithm: bcrypt with cost factor 12 (via passlib)
• Never log passwords (sanitize logs, error messages)
• Enforce minimum strength: 8 characters, mixed case, numbers (frontend + backend validation)

Token Security

• Access tokens: Short-lived (15min), stored in memory (SessionStorage acceptable)
• Refresh tokens: Long-lived (30 days), httpOnly cookies or encrypted storage
• Token hashing: Store only hashed refresh tokens in DB (SHA-256)
• Secret rotation: Support multiple SECRET_KEYs for gradual rotation

CSRF Protection

• Stateless JWTs: No CSRF risk (no cookies for auth)
• If using refresh token cookies: Require SameSite=Strict or CSRF token

Rate Limiting

Add rate limiting to auth endpoints:

# Using slowapi (FastAPI rate limiting)
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post("/auth/login")
@limiter.limit("5/minute")  # 5 login attempts per minute per IP
def login(request: Request, ...):
    pass

Frontend Integration Points

Angular Changes Required

1. Auth Service:

   • Update login() to send email + password
   • Store both access_token and refresh_token
   • Implement token refresh logic in HTTP interceptor
   • Add getCurrentUser() to fetch user profile

2. HTTP Interceptor:

   • Inject access_token in Authorization header
   • Catch 401 errors, attempt refresh, retry request
   • If refresh fails, redirect to login

3. Route Guards:

   • Update AuthGuard to check token validity
   • Add RoleGuard for permission-based route protection
   • Fetch user profile on app initialization

4. UI Components:

   • Add user profile menu (email, logout button)
   • Show project role badges (owner/editor/viewer)
   • Add project sharing dialog (invite members)
   • Show editor attribution in history panel ("Edited by Alice")

Sources

Multi-Tenancy Patterns:

• Introduction to Multi-Tenant Design with FastAPI (2026 guide, HIGH confidence)
• GitHub Discussion: How to implement Multi tenancy in FastAPI
• MergeBoard: Multitenancy with FastAPI, SQLAlchemy and PostgreSQL

Authentication & RBAC:

• Building a Modern User Permission Management System with FastAPI, SQLAlchemy 2.0 (HIGH confidence)
• FastAPI RBAC - Full Implementation Tutorial
• JWT in FastAPI, the Secure Way (Refresh Tokens Explained) (Jan 2026)

Email Verification & Invites:

• User Account Verification Via Email - FastAPI Beyond CRUD
• Top 5 authentication solutions for secure FastAPI apps in 2026

Dependency Injection:

• FastAPI Auth with Dependency Injection
• How to Use Dependency Injection in FastAPI (2026)
• Dependency Injection in FastAPI: 2026 Playbook

SQLite Multi-Tenancy:

• Give each of your users their own SQLite database (Database-per-tenant approach)
• Multi-tenancy - High Performance SQLite

Database Migrations:

• How to do the migration in FastAPI
• FastAPI with Async SQLAlchemy, SQLModel, and Alembic

---

Architecture research for: req-planner v2.0 Multi-Tenancy & User Accounts Researched: 2026-03-18 Confidence: HIGH (Context7 unavailable, but official FastAPI docs + SQLAlchemy 2.0 docs + recent 2026 tutorials provide authoritative patterns)