feat: add runtime system settings module

README.md

@@ -44,6 +44,7 @@ This template integrates the best-in-class Python ecosystem tools to provide a s
 - **Support / Ticketing System:** Full ticket lifecycle — open, reply with image attachments, status/priority, admin assignment — fanned out over a **Redis-backed WebSocket bus** so the ticket thread, the admin queue, and per-user feeds update live across workers. See [Support System & Realtime](#-support-ticketing-system--realtime).
 - **Notification System:** Persistent in-app notifications (`notification` table) layered on the same realtime bus — domain events (a support reply, a ticket status change, an RBAC grant change) call a single `notify()` use case that stores the row **and** pushes it to the recipient's `notifications:{id}` feed, so offline users still find it in their inbox on the next fetch. Notification copy is stored as a stable machine `type` + `data` payload (no human text), so the frontend renders it in any locale. REST covers paginated listing (`unread_only` filter), the unread badge count, and mark-(all-)read. See [Notification System](#-notification-system).
 - **Broadcast / Announcements:** Admins compose an announcement — from a reusable **template catalog** (typed variables: localized datetimes + per-language text) or **custom** per-language text — and fan it out to an audience (`all` / `active` / by `role`). Each recipient gets a persistent notification via a chunked `notify_many` use case (bulk insert + realtime), and optionally a **multilingual email** sent in throttled batches by an `arq` job. Content is stored language-agnostically so it renders in each user's locale; datetimes are stored UTC. A `show_banner` flag exposes the latest as a top banner (`GET /announcements/active`). Gated by `broadcast:read` / `broadcast:write`.
+- **System Settings:** Runtime-editable, **registry-backed** key-value settings — a typed code registry (`core/settings_registry.py`) is the single source of truth (type, default, category, `is_public`, validation) and the `system_setting` table stores **only admin overrides**, so a missing row falls back to the env/registry default. An unauthenticated `GET /settings/public` exposes only `is_public` settings (site name, logo URL, maintenance flag, …) for the login / maintenance screens; admins read & patch all via RBAC (`system_settings:read` / `system_settings:write`). Flags are wired to real behaviour: a **maintenance-mode middleware** 503s non-admins, and registration / support / max-upload-size are enforced from the settings. See [System Settings](#-system-settings).
 - **Session / Device Management:** Every login opens a `user_session` row whose id rides in both tokens as the JWT `sid` claim, so a user can list their active devices and revoke any one (or all but the current). Refresh **rotates** the session's `jti` and detects **replay** — a stale refresh token revokes the whole session as compromised; logout and password change revoke sessions too. Revocation kills a session's still-valid access tokens instantly via a Redis `sid` flag, and broadcasts `sessions_revoked` so open tabs drop to login live. Admins can list/terminate a user's sessions, gated by the `users:sessions` permission. A nightly arq job purges stale rows. See [Session Management](#-session--device-management).
 - **First Superadmin Seed:** On startup a **root superadmin** (`is_root_superadmin`) is created from `FIRST_SUPERUSER` / `FIRST_SUPERUSER_PASSWORD` if none exists (a matching existing account is promoted instead; older deployments get their oldest superadmin flagged as root) — guaranteeing a root account.
 - **Tooling:** [uv](https://docs.astral.sh/uv/) for blazing-fast package management, and [Ruff](https://docs.astral.sh/ruff/) for linting and formatting.
@@ -404,6 +405,24 @@ Persistent, per-user in-app notifications built on the same realtime bus as supp
 
 ---
 
+## ⚙️ System Settings
+
+Runtime-editable, site-wide settings an admin can change without a redeploy (site name, logo, maintenance mode, feature toggles, limits).
+
+**Registry is the single source of truth.** `core/settings_registry.py` defines every setting as a typed, immutable `SettingDefinition` — `value_type` (`bool` / `int` / `string`), `default`, `category`, `is_public`, and an optional validator. The `system_setting` table stores **only admin-overridden values** (`key` → serialised `value`); a setting with no row falls back to its registry default. Several defaults are seeded from the environment at import (`site_name` ← `PROJECT_NAME`, `default_locale` ← `DEFAULT_LANGUAGE`, `max_upload_size_mb` ← `MAX_UPLOAD_SIZE_BYTES`), so settings are initialised from env yet remain runtime-editable. Adding a setting is one registry entry — **no migration**.
+
+| Method & path | Auth | Purpose |
+| --- | --- | --- |
+| `GET /settings/public` | **public** | Only `is_public` settings as `key → typed value` (site name, logo, maintenance flag, …); powers the login & maintenance screens before auth. |
+| `GET /admin/system-settings` | `system_settings:read` | Every setting with metadata (type, category, current value, last editor). |
+| `PATCH /admin/system-settings/{key}` | `system_settings:write` | Update one setting; validated against the registry (type + validator) and audited. |
+
+**Typed access (`use_cases/settings.py`).** `get_bool` / `get_int` / `get_str` resolve a setting to its typed value (DB override → registry/env default). They live in `use_cases/` so any service can read a setting without a service-to-service call.
+
+**Behavioural bindings.** Settings drive real behaviour: a `maintenance_mode_middleware` returns **503** to non-admins while maintenance is on (auth, health and `/settings/public` stay open so admins can sign in); `registration_enabled` gates registration, `support_enabled` gates new tickets, and `max_upload_size_mb` caps uploads. The logo is uploaded through the normal file pipeline (Cloudinary + `file` table) under the `branding_logo` category and its URL is stored in `logo_url`.
+
+---
+
 ## 🔒 Session / Device Management
 
 Per-device login sessions backed by the `user_session` table, so a user can see where they're signed in and revoke access remotely. Each login mints an access + refresh pair carrying the session id as the JWT `sid` claim; the row records the device user-agent (parsed into browser/OS at the schema layer — the raw IP stays server-side and is never returned).

app/alembic/versions/a2290a6fbc7f_add_system_setting_model.py

@@ -0,0 +1,42 @@
+"""add system_setting model
+
+Revision ID: a2290a6fbc7f
+Revises: 1c21a0056027
+Create Date: 2026-06-16 06:00:02.246411
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a2290a6fbc7f"
+down_revision: str | Sequence[str] | None = "1c21a0056027"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "system_setting",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("key", sa.String(length=100), nullable=False),
+        sa.Column("value", sa.Text(), nullable=False),
+        sa.Column("updated_by", sa.Uuid(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["updated_by"], ["user.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("key"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("system_setting")
+    # ### end Alembic commands ###

app/api/deps.py

@@ -195,6 +195,41 @@ async def get_current_session_id(
 CurrentSessionId = Annotated[uuid.UUID | None, Depends(get_current_session_id)]
 
 
+async def resolve_optional_user(request: Request, db: AsyncSession) -> User | None:
+    """Resolve the JWT to a User if present and valid, else None (non-raising).
+
+    A plain helper — NOT a FastAPI dependency — so the maintenance middleware can
+    call it with a manually-opened session. It reads the token straight from the
+    cookie or ``Authorization`` header rather than via ``OAuth2PasswordBearer``,
+    which only resolves inside the HTTP dependency flow (and breaks on WebSocket
+    routes). Mirrors ``get_ws_user`` for HTTP.
+    """
+    token = request.cookies.get("access_token")
+    if not token:
+        auth_header = request.headers.get("Authorization", "")
+        if auth_header.startswith("Bearer "):
+            token = auth_header[len("Bearer ") :]
+    if not token:
+        return None
+    try:
+        if await is_token_blacklisted(token):
+            return None
+        claims = decode_token_payload(token)
+        if claims is None:
+            return None
+        sid = claims.get("sid")
+        if sid and await is_session_revoked(sid):
+            return None
+        token_data = TokenPayload(sub=claims["sub"], sid=sid, jti=claims.get("jti"))
+    except (ValidationError, ValueError):
+        return None
+    user = await get_user_by_id(db, user_id=uuid.UUID(token_data.sub))
+    # Mirror get_ws_user: a deactivated account (even with a still-valid token)
+    # must not pass — otherwise a suspended/deactivated admin could slip through
+    # the maintenance gate.
+    return user if user and user.is_active else None
+
+
 async def user_has_permission(
     session: AsyncSession, user: User, permission: Permission
 ) -> bool:

app/api/main.py

@@ -8,6 +8,7 @@
     health,
     notifications,
     support,
+    system_settings,
     users,
 )
 
@@ -23,5 +24,6 @@
     announcements.router, prefix="/announcements", tags=["announcements"]
 )
 api_router.include_router(support.router, prefix="/support", tags=["support"])
+api_router.include_router(system_settings.router, prefix="/settings", tags=["settings"])
 api_router.include_router(admin.router, prefix="/admin", tags=["admin"])
 api_router.include_router(files.router, tags=["files"])

app/api/middleware.py

@@ -0,0 +1,73 @@
+"""Maintenance-mode HTTP middleware.
+
+Lives in the ``api`` layer (not ``core``) because it depends on auth/session
+helpers (``resolve_optional_user``, ``get_db``); having ``core`` import from
+``api`` would invert the layered dependency direction.
+"""
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from app.api.deps import get_db, resolve_optional_user
+from app.core.config import settings
+from app.core.messages.error_message import ErrorMessages
+from app.schemas.user import SystemRole
+from app.use_cases.settings import get_bool
+
+# API paths that stay reachable while maintenance mode is on, so admins can
+# still authenticate and the frontend can render the maintenance screen.
+_MAINTENANCE_EXEMPT: tuple[str, ...] = (
+    f"{settings.API_V1_STR}/health",
+    f"{settings.API_V1_STR}/auth/login",
+    f"{settings.API_V1_STR}/auth/refresh",
+    f"{settings.API_V1_STR}/auth/logout",
+    f"{settings.API_V1_STR}/settings/public",
+)
+
+
+def register_maintenance_middleware(app: FastAPI) -> None:
+    """Return 503 for non-admins while maintenance mode is on.
+
+    Only API paths are gated. Auth (login/refresh/logout), health, and the
+    public settings endpoint stay open so admins can sign in and the frontend
+    can render the maintenance screen; admins and superadmins pass through.
+    CORS preflight (OPTIONS) is never gated. Runs as HTTP middleware rather than
+    a router dependency so WebSocket routes are untouched.
+    """
+
+    @app.middleware("http")
+    async def maintenance_mode_middleware(request: Request, call_next):
+        """Block non-admins with 503 while maintenance mode is on."""
+        path = request.url.path
+        if (
+            request.method == "OPTIONS"
+            or not path.startswith(settings.API_V1_STR)
+            or path.startswith(_MAINTENANCE_EXEMPT)
+        ):
+            return await call_next(request)
+
+        # Resolve the session through ``get_db`` honouring ``dependency_overrides``
+        # so tests reuse their loop-bound session; production gets the real one.
+        # The session is closed before ``call_next`` so the connection is not held
+        # for the downstream request.
+        db_factory = request.app.dependency_overrides.get(get_db, get_db)
+        agen = db_factory()
+        session = await agen.__anext__()
+        try:
+            maintenance_on = await get_bool(session, "maintenance_mode")
+            user = (
+                await resolve_optional_user(request, session)
+                if maintenance_on
+                else None
+            )
+        finally:
+            await agen.aclose()
+
+        if not maintenance_on:
+            return await call_next(request)
+        if user and user.role in (SystemRole.ADMIN, SystemRole.SUPERADMIN):
+            return await call_next(request)
+        return JSONResponse(
+            status_code=503,
+            content={"success": False, "error": ErrorMessages.MAINTENANCE_MODE},
+        )

app/api/routes/admin/__init__.py

@@ -13,6 +13,7 @@
     files,
     stats,
     support,
+    system_settings,
     users,
 )
 
@@ -24,3 +25,4 @@
 router.include_router(activities.router)
 router.include_router(stats.router)
 router.include_router(broadcast.router)
+router.include_router(system_settings.router)

app/api/routes/admin/system_settings.py

@@ -0,0 +1,63 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Request
+
+from app.api.decorators import audit_unexpected_failure
+from app.api.deps import SessionDep, require_permission
+from app.core.rate_limit import rate_limit_authenticated
+from app.models.user import User
+from app.schemas.admin_permission import Permission
+from app.schemas.system_settings import (
+    SettingsListResponse,
+    SettingUpdate,
+    SettingUpdateResponse,
+)
+from app.schemas.user_activity import ActivityType, ResourceType
+from app.services.system_settings import (
+    list_all_settings_service,
+    update_setting_service,
+)
+
+router = APIRouter()
+
+AdminSettingsRead = Annotated[
+    User, Depends(require_permission(Permission.SYSTEM_SETTINGS_READ))
+]
+AdminSettingsWrite = Annotated[
+    User, Depends(require_permission(Permission.SYSTEM_SETTINGS_WRITE))
+]
+
+
+@router.get("/system-settings", response_model=SettingsListResponse)
+@audit_unexpected_failure(
+    activity_type=ActivityType.READ,
+    resource_type=ResourceType.SYSTEM_SETTINGS,
+    endpoint="/admin/system-settings",
+)
+async def list_system_settings(
+    _request: Request,
+    _admin: AdminSettingsRead,
+    session: SessionDep,
+) -> SettingsListResponse:
+    """List every system setting with its metadata and current value."""
+    return await list_all_settings_service(session)
+
+
+@router.patch("/system-settings/{key}", response_model=SettingUpdateResponse)
+@rate_limit_authenticated("30/minute")
+@audit_unexpected_failure(
+    activity_type=ActivityType.UPDATE,
+    resource_type=ResourceType.SYSTEM_SETTINGS,
+    endpoint="/admin/system-settings/{key}",
+)
+async def update_system_setting(
+    request: Request,
+    admin: AdminSettingsWrite,
+    session: SessionDep,
+    payload: SettingUpdate,
+    key: Annotated[str, Path()],
+) -> SettingUpdateResponse:
+    """Update one system setting's value (RBAC-guarded and audited)."""
+    return await update_setting_service(
+        session=session, key=key, payload=payload, admin=admin, request=request
+    )

app/api/routes/system_settings.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, Request
+
+from app.api.deps import SessionDep
+from app.core.rate_limit import rate_limit_public
+from app.schemas.system_settings import PublicSettingsResponse
+from app.services.system_settings import get_public_settings_service
+
+router = APIRouter()
+
+
+@router.get("/public", response_model=PublicSettingsResponse)
+@rate_limit_public("60/minute")
+async def read_public_settings(
+    request: Request,  # noqa: ARG001 - required in the signature by slowapi
+    session: SessionDep,
+) -> PublicSettingsResponse:
+    """Return public system settings for unauthenticated clients.
+
+    Powers the login screen and maintenance gate before authentication: only
+    settings flagged ``is_public`` in the registry are exposed here.
+    """
+    return await get_public_settings_service(session)

app/core/messages/error_message.py

@@ -82,3 +82,12 @@ class ErrorMessages:
     ANNOUNCEMENT_TEMPLATE_NOT_FOUND = "error.announcement.template_not_found"
     ANNOUNCEMENT_INVALID_CONTENT = "error.announcement.invalid_content"
     ANNOUNCEMENT_INVALID_AUDIENCE = "error.announcement.invalid_audience"
+
+    # System settings
+    SETTING_NOT_FOUND = "error.settings.not_found"
+    SETTING_INVALID_VALUE = "error.settings.invalid_value"
+
+    # System / availability
+    MAINTENANCE_MODE = "error.system.maintenance"
+    REGISTRATION_DISABLED = "error.auth.registration_disabled"
+    SUPPORT_DISABLED = "error.support.disabled"

app/core/messages/success_message.py

@@ -55,3 +55,6 @@ class SuccessMessages:
 
     # Announcements / broadcast
     BROADCAST_SENT = "success.announcement.sent"
+
+    # System settings
+    SETTINGS_UPDATED = "success.settings.updated"