diff --git a/app/api/admin_users.py b/app/api/admin_users.py
index f80d463f..47c9d042 100644
--- a/app/api/admin_users.py
+++ b/app/api/admin_users.py
@@ -59,6 +59,11 @@ class UserProfileUpsert(BaseModel):
subscription_billing_cycle: str = Field(default="monthly", pattern="^(monthly|yearly)$")
subscription_period_start: datetime | None = None
allow_overage: bool = False
+ is_complimentary: bool = Field(
+ default=False,
+ description="When True the user is on a complimentary (uncharged) plan — they keep all tier "
+ "quota benefits but are never billed via Stripe.",
+ )
class UserProfileResponse(BaseModel):
@@ -74,6 +79,7 @@ class UserProfileResponse(BaseModel):
subscription_billing_cycle: str
subscription_period_start: str | None
allow_overage: bool
+ is_complimentary: bool
created_at: str | None
updated_at: str | None
@@ -92,6 +98,7 @@ class UserSummary(BaseModel):
subscription_billing_cycle: str | None
subscription_period_start: str | None
allow_overage: bool
+ is_complimentary: bool
profile_id: int | None
document_count: int
last_upload: str | None
@@ -121,6 +128,7 @@ def _profile_to_dict(profile: UserProfile) -> dict[str, Any]:
if profile.subscription_period_start
else None,
"allow_overage": bool(profile.allow_overage),
+ "is_complimentary": bool(profile.is_complimentary),
"created_at": profile.created_at.isoformat() if profile.created_at else None,
"updated_at": profile.updated_at.isoformat() if profile.updated_at else None,
}
@@ -194,6 +202,7 @@ def list_users(
if (profile and profile.subscription_period_start)
else None,
"allow_overage": bool(profile.allow_overage) if profile else False,
+ "is_complimentary": bool(profile.is_complimentary) if profile else False,
"profile_id": profile.id if profile else None,
"document_count": doc_row.doc_count if doc_row else 0,
"last_upload": doc_row.last_upload.isoformat() if (doc_row and doc_row.last_upload) else None,
@@ -235,6 +244,7 @@ def get_user(user_id: str, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
if (profile and profile.subscription_period_start)
else None,
"allow_overage": bool(profile.allow_overage) if profile else False,
+ "is_complimentary": bool(profile.is_complimentary) if profile else False,
"profile_id": profile.id if profile else None,
"document_count": doc_count,
"last_upload": last_upload,
@@ -265,6 +275,7 @@ def upsert_user_profile(
profile.subscription_billing_cycle = body.subscription_billing_cycle
profile.subscription_period_start = body.subscription_period_start
profile.allow_overage = body.allow_overage
+ profile.is_complimentary = body.is_complimentary
if body.subscription_tier is not None:
from app.utils.subscription import TIERS
diff --git a/app/auth.py b/app/auth.py
index e0399915..99b67647 100644
--- a/app/auth.py
+++ b/app/auth.py
@@ -127,18 +127,35 @@ async def oauth_login(request: Request):
return await oauth.authentik.authorize_redirect(request, redirect_uri)
-def _ensure_user_profile(db: Session, user_data: dict) -> None:
- """Create a UserProfile row for *user_data* if one does not yet exist.
+def _ensure_user_profile(db: Session, user_data: dict, is_admin: bool = False) -> None:
+ """Create or update a UserProfile row for *user_data*.
Uses the same identifier priority as ``get_current_owner_id`` (sub →
preferred_username → email → id) so that the profile's ``user_id`` matches
``FileRecord.owner_id`` for every document the user uploads.
- If a profile already exists it is left unchanged; only missing profiles
- are created so that admin-managed settings (tier, limits, etc.) are
- preserved across logins.
+ For regular users, an existing profile is left unchanged so that
+ admin-managed settings (tier, limits, etc.) are preserved across logins.
+
+ For admin users (*is_admin=True*) the following rules apply:
+ - If no profile exists: one is created with the highest subscription tier,
+ ``is_complimentary=True``, and ``onboarding_completed=True`` so that
+ admins skip the first-time setup wizard.
+ - If a profile already exists: ``is_complimentary`` is set to ``True``
+ and, when the current tier is ``"free"``, the tier is upgraded to the
+ highest available plan. Other admin-managed settings are left intact.
+
+ Args:
+ db: Active database session.
+ user_data: Mapping of user attributes as returned by the OAuth provider
+ or built by :func:`app.utils.local_auth.build_session_user`.
+ is_admin: When ``True``, apply admin-specific defaults on first login
+ and ensure the complimentary flag is always set.
"""
from app.models import UserProfile
+ from app.utils.subscription import TIER_ORDER
+
+ highest_tier = TIER_ORDER[-1]
user_id = (
user_data.get("sub") or user_data.get("preferred_username") or user_data.get("email") or user_data.get("id")
@@ -151,13 +168,41 @@ def _ensure_user_profile(db: Session, user_data: dict) -> None:
existing = db.query(UserProfile).filter(UserProfile.user_id == user_id).first()
if existing is None:
display_name = user_data.get("name") or user_data.get("preferred_username") or user_data.get("email")
- profile = UserProfile(user_id=user_id, display_name=display_name)
+ profile = UserProfile(
+ user_id=user_id,
+ display_name=display_name,
+ subscription_tier=highest_tier if is_admin else "free",
+ is_complimentary=is_admin,
+ onboarding_completed=is_admin,
+ )
db.add(profile)
db.commit()
- logger.info("Auto-created UserProfile for user_id=%s", user_id)
+ logger.info(
+ "Auto-created UserProfile for user_id=%s (admin=%s, tier=%s)",
+ user_id,
+ is_admin,
+ highest_tier if is_admin else "free",
+ )
+ elif is_admin:
+ # Ensure existing admin profiles always have complimentary flag set.
+ # Also upgrade from free tier to highest if still on default.
+ changed = False
+ if not existing.is_complimentary:
+ existing.is_complimentary = True
+ changed = True
+ if (existing.subscription_tier or "free") == "free":
+ existing.subscription_tier = highest_tier
+ changed = True
+ if changed:
+ db.commit()
+ logger.info(
+ "Updated admin UserProfile for user_id=%s (complimentary=True, tier=%s)",
+ user_id,
+ existing.subscription_tier,
+ )
except Exception:
db.rollback()
- logger.exception("Failed to auto-create UserProfile for user_id=%s", user_id)
+ logger.exception("Failed to auto-create/update UserProfile for user_id=%s", user_id)
async def oauth_callback(request: Request, db: Session = Depends(get_db)):
@@ -193,7 +238,7 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)):
request.session["user"] = user_data
# Auto-create or update UserProfile so the user appears in admin user management
- _ensure_user_profile(db, user_data)
+ _ensure_user_profile(db, user_data, is_admin=is_admin)
# Log the successful authentication
logger.info("[SECURITY] OAUTH_LOGIN_SUCCESS user=%s admin=%s", user_data.get("email", "unknown"), is_admin)
@@ -251,6 +296,7 @@ async def auth(request: Request, db: Session = Depends(get_db)):
user_data = _build_session_user(local_user)
request.session["user"] = user_data
logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", local_user.email)
+ _ensure_user_profile(db, user_data, is_admin=bool(local_user.is_admin))
profile = db.query(_UserProfile).filter(_UserProfile.user_id == local_user.email).first()
if profile and not profile.onboarding_completed:
post_onboarding = request.session.pop("redirect_after_login", "/upload")
@@ -261,7 +307,7 @@ async def auth(request: Request, db: Session = Depends(get_db)):
# --- Admin credentials (always available as a fallback / single-user mode) ---
if username == settings.admin_username and password == settings.admin_password:
- request.session["user"] = {
+ admin_user_data = {
"id": "admin",
"name": "Administrator",
"email": f"{username}@local.docuelevate",
@@ -269,7 +315,9 @@ async def auth(request: Request, db: Session = Depends(get_db)):
"picture": "/static/images/default-avatar.svg",
"is_admin": True,
}
+ request.session["user"] = admin_user_data
logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", username)
+ _ensure_user_profile(db, admin_user_data, is_admin=True)
redirect_url = request.session.pop("redirect_after_login", "/upload")
return RedirectResponse(url=redirect_url, status_code=302)
else:
diff --git a/app/models.py b/app/models.py
index 1f160041..5ea34eb9 100644
--- a/app/models.py
+++ b/app/models.py
@@ -238,6 +238,10 @@ class UserProfile(Base):
subscription_period_start = Column(DateTime(timezone=True), nullable=True)
allow_overage = Column(Boolean, nullable=False, default=False, server_default="0")
+ # When True, the user is on a complimentary (uncharged) plan — they keep all tier
+ # quota benefits but are never billed via Stripe. Automatically set for admin users.
+ is_complimentary = Column(Boolean, nullable=False, default=False, server_default="0")
+
# Onboarding tracking (added in migration 017)
onboarding_completed = Column(Boolean, nullable=False, default=False, server_default="0")
onboarding_completed_at = Column(DateTime(timezone=True), nullable=True)
diff --git a/docs/SubscriptionTiers.md b/docs/SubscriptionTiers.md
index d013fff7..b895e762 100644
--- a/docs/SubscriptionTiers.md
+++ b/docs/SubscriptionTiers.md
@@ -55,6 +55,35 @@ When a user's `subscription_billing_cycle` is set to `yearly`:
Setting `UserProfile.allow_overage = True` bypasses monthly quota checks entirely for that user. Usage is still tracked so future billing integrations can charge retroactively. This field is not yet exposed in the admin UI.
+## is_complimentary Flag (Complimentary Plans)
+
+Setting `UserProfile.is_complimentary = True` marks a user as being on a **complimentary (uncharged) plan**. The user retains all quota benefits of their assigned subscription tier but is **never billed via Stripe**. This is useful for:
+
+- **Admin accounts** — automatically set on every admin user profile at login time.
+- **Gifted access** — granting full plan benefits to partners, testers, or sponsored users.
+
+### Admin Auto-Provisioning
+
+When an admin user logs in for the first time (via OAuth, local account, or the built-in admin credentials), DocuElevate automatically:
+
+1. Creates a `UserProfile` row if one does not already exist.
+2. Assigns the **highest available subscription tier** (currently `business`).
+3. Sets `is_complimentary = True` so the account is never billed.
+4. Sets `onboarding_completed = True` so admins skip the first-time setup wizard.
+
+On subsequent logins for existing admin profiles:
+- `is_complimentary` is ensured to be `True`.
+- If the profile was still on the `free` tier it is upgraded to the highest tier.
+- All other admin-managed settings (custom limits, notes, etc.) are preserved.
+
+### Managing via Admin UI
+
+The **User Management** page (`/admin/users`) shows a green gift icon (🎁) next to the plan badge for any user with `is_complimentary = True`. The toggle is available in the user edit modal under **Billing**.
+
+### API Field
+
+`is_complimentary` is exposed in the `PUT /api/admin/users/{user_id}` body and in all user detail responses.
+
## Plan Designer
Navigate to `/admin/plans` (admin only) to:
diff --git a/frontend/templates/admin_users.html b/frontend/templates/admin_users.html
index 84ed2042..4b2b2d18 100644
--- a/frontend/templates/admin_users.html
+++ b/frontend/templates/admin_users.html
@@ -140,6 +140,14 @@
>
+
+
+ Complimentary plan
+
@@ -375,6 +383,25 @@
+
+
+
+
+
+
@@ -472,6 +499,7 @@ function adminUsersApp() {
subscription_tier: 'free',
subscription_billing_cycle: 'monthly',
subscription_period_start: null,
+ is_complimentary: false,
},
// Delete modal
@@ -524,7 +552,7 @@ function adminUsersApp() {
openCreateModal() {
this.isCreate = true;
this.modalTitle = 'Add User Profile';
- this.form = { user_id: '', display_name: '', daily_upload_limit: null, notes: '', is_blocked: false, subscription_tier: 'free', subscription_billing_cycle: 'monthly', subscription_period_start: null };
+ this.form = { user_id: '', display_name: '', daily_upload_limit: null, notes: '', is_blocked: false, subscription_tier: 'free', subscription_billing_cycle: 'monthly', subscription_period_start: null, is_complimentary: false };
this.modalOpen = true;
},
@@ -541,6 +569,7 @@ function adminUsersApp() {
subscription_tier: user.subscription_tier || 'free',
subscription_billing_cycle: user.subscription_billing_cycle || 'monthly',
subscription_period_start: user.subscription_period_start ? user.subscription_period_start.substring(0, 10) : null,
+ is_complimentary: !!user.is_complimentary,
};
this.modalOpen = true;
},
@@ -555,6 +584,7 @@ function adminUsersApp() {
notes: this.form.notes || null,
is_blocked: !!this.form.is_blocked,
subscription_tier: this.form.subscription_tier || 'free',
+ is_complimentary: !!this.form.is_complimentary,
};
const uid = encodeURIComponent(this.form.user_id);
const resp = await fetch(`/api/admin/users/${uid}`, {
diff --git a/migrations/versions/019_add_is_complimentary.py b/migrations/versions/019_add_is_complimentary.py
new file mode 100644
index 00000000..a66b355c
--- /dev/null
+++ b/migrations/versions/019_add_is_complimentary.py
@@ -0,0 +1,30 @@
+"""Add is_complimentary column to user_profiles
+
+Revision ID: 019_add_is_complimentary
+Revises: 018_add_local_users_and_billing
+Create Date: 2026-03-07
+
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "019_add_is_complimentary"
+down_revision: Union[str, None] = "018_add_local_users_and_billing"
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+ """Add is_complimentary column to user_profiles."""
+ op.add_column(
+ "user_profiles",
+ sa.Column("is_complimentary", sa.Boolean(), nullable=False, server_default="0"),
+ )
+
+
+def downgrade() -> None:
+ """Remove is_complimentary column from user_profiles."""
+ op.drop_column("user_profiles", "is_complimentary")
diff --git a/tests/test_admin_users.py b/tests/test_admin_users.py
index e889a704..f961b065 100644
--- a/tests/test_admin_users.py
+++ b/tests/test_admin_users.py
@@ -477,3 +477,189 @@ class TestUserProfileModel:
with pytest.raises(IntegrityError):
au_session.commit()
au_session.rollback()
+
+
+# ---------------------------------------------------------------------------
+# Complimentary plan tests
+# ---------------------------------------------------------------------------
+
+
+class TestComplimentaryPlan:
+ """Tests for the is_complimentary field and admin auto-creation logic."""
+
+ @pytest.mark.unit
+ def test_create_profile_with_complimentary_flag(self, au_client, au_session):
+ """PUT can create a profile with is_complimentary=True."""
+ resp = au_client.put(
+ "/api/admin/users/comp@example.com",
+ json={"subscription_tier": "business", "is_complimentary": True, "is_blocked": False},
+ )
+ assert resp.status_code == 200
+ data = resp.json()
+ assert data["is_complimentary"] is True
+ assert data["subscription_tier"] == "business"
+
+ profile = au_session.query(UserProfile).filter_by(user_id="comp@example.com").first()
+ assert profile is not None
+ assert profile.is_complimentary is True
+
+ @pytest.mark.unit
+ def test_update_profile_set_complimentary(self, au_client, au_session):
+ """PUT can toggle is_complimentary on an existing profile."""
+ _make_profile(au_session, "toggle@example.com", is_complimentary=False)
+
+ resp = au_client.put(
+ "/api/admin/users/toggle@example.com",
+ json={"is_blocked": False, "is_complimentary": True},
+ )
+ assert resp.status_code == 200
+ assert resp.json()["is_complimentary"] is True
+
+ @pytest.mark.unit
+ def test_list_users_includes_complimentary_field(self, au_client, au_session):
+ """GET /api/admin/users/ returns is_complimentary per user."""
+ _make_profile(au_session, "complist@example.com", is_complimentary=True)
+
+ resp = au_client.get("/api/admin/users/")
+ assert resp.status_code == 200
+ users = {u["user_id"]: u for u in resp.json()["users"]}
+ assert "complist@example.com" in users
+ assert users["complist@example.com"]["is_complimentary"] is True
+
+ @pytest.mark.unit
+ def test_get_user_includes_complimentary_field(self, au_client, au_session):
+ """GET /api/admin/users/ returns is_complimentary in profile."""
+ _make_profile(au_session, "getcomp@example.com", is_complimentary=True, subscription_tier="business")
+
+ resp = au_client.get("/api/admin/users/getcomp%40example.com")
+ assert resp.status_code == 200
+ data = resp.json()
+ assert data["is_complimentary"] is True
+ assert data["profile"]["is_complimentary"] is True
+
+ @pytest.mark.unit
+ def test_complimentary_defaults_to_false(self, au_client, au_session):
+ """Newly created profiles have is_complimentary=False by default."""
+ resp = au_client.put(
+ "/api/admin/users/nocomp@example.com",
+ json={"is_blocked": False},
+ )
+ assert resp.status_code == 200
+ assert resp.json()["is_complimentary"] is False
+
+ @pytest.mark.unit
+ def test_profile_model_complimentary_column(self, au_session):
+ """UserProfile model stores is_complimentary correctly."""
+ profile = UserProfile(user_id="modelcomp@example.com", is_complimentary=True)
+ au_session.add(profile)
+ au_session.commit()
+ au_session.refresh(profile)
+ assert profile.is_complimentary is True
+
+
+# ---------------------------------------------------------------------------
+# _ensure_user_profile admin auto-creation tests
+# ---------------------------------------------------------------------------
+
+
+class TestEnsureUserProfileAdmin:
+ """Tests for _ensure_user_profile admin-specific behaviour."""
+
+ @pytest.mark.unit
+ def test_admin_login_creates_highest_tier_profile(self, au_session):
+ """Admin first login creates a profile with the highest subscription tier."""
+ from app.auth import _ensure_user_profile
+ from app.utils.subscription import TIER_ORDER
+
+ user_data = {
+ "preferred_username": "admin",
+ "email": "admin@local.docuelevate",
+ "name": "Administrator",
+ "is_admin": True,
+ }
+ _ensure_user_profile(au_session, user_data, is_admin=True)
+
+ # user_id uses preferred_username (sub not provided)
+ profile = au_session.query(UserProfile).filter_by(user_id="admin").first()
+ assert profile is not None
+ assert profile.subscription_tier == TIER_ORDER[-1]
+ assert profile.is_complimentary is True
+ assert profile.onboarding_completed is True
+
+ @pytest.mark.unit
+ def test_regular_user_login_creates_free_profile(self, au_session):
+ """Regular user login creates a profile with the free tier."""
+ from app.auth import _ensure_user_profile
+
+ user_data = {
+ "preferred_username": "regular",
+ "email": "user@example.com",
+ "name": "Regular User",
+ }
+ _ensure_user_profile(au_session, user_data, is_admin=False)
+
+ # user_id uses preferred_username (sub not provided)
+ profile = au_session.query(UserProfile).filter_by(user_id="regular").first()
+ assert profile is not None
+ assert profile.subscription_tier == "free"
+ assert profile.is_complimentary is False
+
+ @pytest.mark.unit
+ def test_admin_login_sets_complimentary_on_existing_profile(self, au_session):
+ """Existing admin profile gets is_complimentary=True on login."""
+ existing = UserProfile(user_id="existadmin", is_complimentary=False, subscription_tier="starter")
+ au_session.add(existing)
+ au_session.commit()
+
+ from app.auth import _ensure_user_profile
+
+ user_data = {"preferred_username": "existadmin", "email": "ea@example.com"}
+ _ensure_user_profile(au_session, user_data, is_admin=True)
+
+ au_session.refresh(existing)
+ assert existing.is_complimentary is True
+
+ @pytest.mark.unit
+ def test_admin_login_does_not_downgrade_existing_tier(self, au_session):
+ """Existing admin profile with a paid tier keeps that tier on re-login."""
+ from app.auth import _ensure_user_profile
+ from app.utils.subscription import TIER_ORDER
+
+ highest = TIER_ORDER[-1]
+ existing = UserProfile(user_id="toptieradmin", is_complimentary=False, subscription_tier=highest)
+ au_session.add(existing)
+ au_session.commit()
+
+ user_data = {"preferred_username": "toptieradmin", "email": "tt@example.com"}
+ _ensure_user_profile(au_session, user_data, is_admin=True)
+
+ au_session.refresh(existing)
+ assert existing.subscription_tier == highest
+ assert existing.is_complimentary is True
+
+ @pytest.mark.unit
+ def test_admin_login_upgrades_free_tier_on_existing_profile(self, au_session):
+ """Existing admin profile on free tier gets upgraded to highest tier."""
+ from app.auth import _ensure_user_profile
+ from app.utils.subscription import TIER_ORDER
+
+ existing = UserProfile(user_id="freeadmin", is_complimentary=False, subscription_tier="free")
+ au_session.add(existing)
+ au_session.commit()
+
+ user_data = {"preferred_username": "freeadmin", "email": "fa@example.com"}
+ _ensure_user_profile(au_session, user_data, is_admin=True)
+
+ au_session.refresh(existing)
+ assert existing.subscription_tier == TIER_ORDER[-1]
+ assert existing.is_complimentary is True
+
+ @pytest.mark.unit
+ def test_ensure_user_profile_no_identifier_logs_warning(self, au_session):
+ """_ensure_user_profile logs a warning when no stable user id is present."""
+ from app.auth import _ensure_user_profile
+
+ _ensure_user_profile(au_session, {}, is_admin=False)
+ # No profile should have been created
+ count = au_session.query(UserProfile).count()
+ assert count == 0
diff --git a/tests/test_auth_module.py b/tests/test_auth_module.py
index 7cf30187..d08830ff 100644
--- a/tests/test_auth_module.py
+++ b/tests/test_auth_module.py
@@ -278,8 +278,10 @@ class TestAuthEndpoint:
mock_form_data = {"username": "admin", "password": "secret123"}
mock_request.form = AsyncMock(return_value=mock_form_data)
mock_request.session = {}
+ mock_db = MagicMock()
- result = await auth(mock_request)
+ with patch("app.auth._ensure_user_profile"):
+ result = await auth(mock_request, db=mock_db)
# Verify redirect to upload page
assert isinstance(result, RedirectResponse)
@@ -305,8 +307,9 @@ class TestAuthEndpoint:
mock_form_data = {"username": "admin", "password": "wrong_password"}
mock_request.form = AsyncMock(return_value=mock_form_data)
mock_request.session = {}
+ mock_db = MagicMock()
- result = await auth(mock_request)
+ result = await auth(mock_request, db=mock_db)
# Verify redirect to login with error
assert isinstance(result, RedirectResponse)
@@ -327,8 +330,10 @@ class TestAuthEndpoint:
mock_form_data = {"username": "admin", "password": "secret123"}
mock_request.form = AsyncMock(return_value=mock_form_data)
mock_request.session = {"redirect_after_login": "/protected/page"}
+ mock_db = MagicMock()
- result = await auth(mock_request)
+ with patch("app.auth._ensure_user_profile"):
+ result = await auth(mock_request, db=mock_db)
# Verify redirect to saved URL
assert isinstance(result, RedirectResponse)