diff --git a/README.md b/README.md
index b28e3a5..b379c79 100644
--- a/README.md
+++ b/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).
diff --git a/app/alembic/versions/a2290a6fbc7f_add_system_setting_model.py b/app/alembic/versions/a2290a6fbc7f_add_system_setting_model.py
new file mode 100644
index 0000000..b77e945
--- /dev/null
+++ b/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 ###
diff --git a/app/api/deps.py b/app/api/deps.py
index e0e24e6..a76da4e 100644
--- a/app/api/deps.py
+++ b/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:
diff --git a/app/api/main.py b/app/api/main.py
index 1ccd68e..73df3ef 100644
--- a/app/api/main.py
+++ b/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"])
diff --git a/app/api/middleware.py b/app/api/middleware.py
new file mode 100644
index 0000000..8a03df3
--- /dev/null
+++ b/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},
+ )
diff --git a/app/api/routes/admin/__init__.py b/app/api/routes/admin/__init__.py
index 2567f04..dcb6be6 100644
--- a/app/api/routes/admin/__init__.py
+++ b/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)
diff --git a/app/api/routes/admin/system_settings.py b/app/api/routes/admin/system_settings.py
new file mode 100644
index 0000000..925c87f
--- /dev/null
+++ b/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
+ )
diff --git a/app/api/routes/system_settings.py b/app/api/routes/system_settings.py
new file mode 100644
index 0000000..708591f
--- /dev/null
+++ b/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)
diff --git a/app/core/messages/error_message.py b/app/core/messages/error_message.py
index 9cf5bc2..3256f33 100644
--- a/app/core/messages/error_message.py
+++ b/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"
diff --git a/app/core/messages/success_message.py b/app/core/messages/success_message.py
index 80ab887..d57eaa2 100644
--- a/app/core/messages/success_message.py
+++ b/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"
diff --git a/app/core/settings_registry.py b/app/core/settings_registry.py
new file mode 100644
index 0000000..3894f86
--- /dev/null
+++ b/app/core/settings_registry.py
@@ -0,0 +1,168 @@
+"""Central, typed registry of system settings.
+
+Single source of truth for every configurable setting: its type, default,
+category, public visibility, and validation. The ``system_setting`` table only
+stores admin-overridden *values* keyed by ``key``; everything else lives here,
+so a setting cannot drift between code and database.
+
+Resolution order at read time is: DB override → env/config default (where a
+definition's default is sourced from :data:`app.core.config.settings`) →
+static fallback below. A missing DB row therefore falls back to the env value,
+so settings are initialised from the environment yet remain editable at runtime.
+"""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from enum import StrEnum
+
+from app.core.config import settings
+
+# Settings only ever hold scalar values; no nested/JSON settings in phase one.
+SettingValue = bool | int | str
+
+
+class SettingValueType(StrEnum):
+ """The concrete Python type a setting's value coerces to."""
+
+ BOOL = "bool"
+ INT = "int"
+ STRING = "string"
+
+
+class SettingCategory(StrEnum):
+ """Logical group a setting belongs to, used to tab the admin UI."""
+
+ SYSTEM = "system"
+ AUTH = "auth"
+ UPLOADS = "uploads"
+ SUPPORT = "support"
+ BRANDING = "branding"
+
+
+@dataclass(frozen=True)
+class SettingDefinition:
+ """Immutable metadata describing one configurable setting."""
+
+ key: str
+ value_type: SettingValueType
+ default: SettingValue
+ category: SettingCategory
+ is_public: bool
+ description: str
+ validate: Callable[[SettingValue], None] | None = None
+
+
+def _validate_locale(value: SettingValue) -> None:
+ """Reject any default locale other than the supported ``tr``/``en``."""
+ if value not in ("tr", "en"):
+ raise ValueError("default_locale must be 'tr' or 'en'")
+
+
+def _validate_upload_size(value: SettingValue) -> None:
+ """Keep the max upload size within a sane 1-100 MB band."""
+ if not isinstance(value, int) or not 1 <= value <= 100:
+ raise ValueError("max_upload_size_mb must be between 1 and 100")
+
+
+def _validate_non_empty(value: SettingValue) -> None:
+ """Reject blank strings for settings that must carry text."""
+ if not isinstance(value, str) or not value.strip():
+ raise ValueError("value must be a non-empty string")
+
+
+_DEFINITIONS: tuple[SettingDefinition, ...] = (
+ SettingDefinition(
+ "maintenance_mode",
+ SettingValueType.BOOL,
+ False,
+ SettingCategory.SYSTEM,
+ True,
+ "When on, only admins can reach the app; everyone else gets 503.",
+ ),
+ SettingDefinition(
+ "registration_enabled",
+ SettingValueType.BOOL,
+ True,
+ SettingCategory.AUTH,
+ True,
+ "When off, new account registration is rejected.",
+ ),
+ SettingDefinition(
+ "max_upload_size_mb",
+ SettingValueType.INT,
+ settings.MAX_UPLOAD_SIZE_BYTES // (1024 * 1024),
+ SettingCategory.UPLOADS,
+ False,
+ "Maximum allowed upload size in megabytes.",
+ validate=_validate_upload_size,
+ ),
+ SettingDefinition(
+ "support_enabled",
+ SettingValueType.BOOL,
+ True,
+ SettingCategory.SUPPORT,
+ True,
+ "When off, users cannot open new support tickets.",
+ ),
+ SettingDefinition(
+ "site_name",
+ SettingValueType.STRING,
+ settings.PROJECT_NAME,
+ SettingCategory.BRANDING,
+ True,
+ "Product/site display name.",
+ validate=_validate_non_empty,
+ ),
+ SettingDefinition(
+ "logo_url",
+ SettingValueType.STRING,
+ "",
+ SettingCategory.BRANDING,
+ True,
+ "Public URL of the site logo (stored on Cloudinary).",
+ ),
+ SettingDefinition(
+ "support_email",
+ SettingValueType.STRING,
+ "support@example.com",
+ SettingCategory.BRANDING,
+ True,
+ "Contact/support email address.",
+ validate=_validate_non_empty,
+ ),
+ SettingDefinition(
+ "default_locale",
+ SettingValueType.STRING,
+ settings.DEFAULT_LANGUAGE,
+ SettingCategory.BRANDING,
+ True,
+ "Default language for new users (tr/en).",
+ validate=_validate_locale,
+ ),
+)
+
+SETTINGS_REGISTRY: dict[str, SettingDefinition] = {d.key: d for d in _DEFINITIONS}
+
+
+def get_definition(key: str) -> SettingDefinition | None:
+ """Return the definition for ``key`` or ``None`` if it is unknown."""
+ return SETTINGS_REGISTRY.get(key)
+
+
+def coerce(definition: SettingDefinition, raw: str) -> SettingValue:
+ """Parse a stored string into its typed value per the definition.
+
+ Raises ``ValueError`` if the stored text does not match the declared type.
+ """
+ if definition.value_type is SettingValueType.BOOL:
+ return raw.lower() == "true"
+ if definition.value_type is SettingValueType.INT:
+ return int(raw)
+ return raw
+
+
+def to_str(value: SettingValue) -> str:
+ """Serialise a typed value to its DB string form."""
+ if isinstance(value, bool):
+ return "true" if value else "false"
+ return str(value)
diff --git a/app/main.py b/app/main.py
index d2ad015..20b5d85 100644
--- a/app/main.py
+++ b/app/main.py
@@ -9,6 +9,7 @@
from app.api.exception_handlers import register_exception_handlers
from app.api.main import api_router
+from app.api.middleware import register_maintenance_middleware
from app.core.config import settings
from app.core.lifespan import lifespan
from app.core.metrics import init_metrics
@@ -28,6 +29,7 @@
register_exception_handlers(app)
register_middleware(app)
+register_maintenance_middleware(app)
app.include_router(api_router, prefix=settings.API_V1_STR)
# OpenTelemetry tracing — no-op unless OTEL_EXPORTER_OTLP_ENDPOINT is set.
diff --git a/app/models/__init__.py b/app/models/__init__.py
index b21f600..d86fb78 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -7,6 +7,7 @@
SupportMessageAttachment,
SupportTicket,
)
+from app.models.system_setting import SystemSetting
from app.models.user import User
from app.models.user_activity import UserActivity
from app.models.user_session import UserSession
@@ -19,6 +20,7 @@
"SupportMessage",
"SupportMessageAttachment",
"SupportTicket",
+ "SystemSetting",
"User",
"UserActivity",
"UserSession",
diff --git a/app/models/system_setting.py b/app/models/system_setting.py
new file mode 100644
index 0000000..1937cc2
--- /dev/null
+++ b/app/models/system_setting.py
@@ -0,0 +1,48 @@
+import uuid
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import DateTime, ForeignKey, String, Text
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.core.db import Base
+from app.utils import utc_now
+
+if TYPE_CHECKING:
+ from app.models.user import User
+
+
+class SystemSetting(Base):
+ """An admin-overridden value for a setting defined in the registry.
+
+ Only settings whose value differs from the registry/env default have a row
+ here; a missing row means the default applies. ``key`` matches a key in
+ ``app.core.settings_registry``; ``value`` is the serialised string form
+ (parsed back to its typed value via the registry). Type, category, default
+ and public visibility are NOT stored here — the registry is their single
+ source of truth, so settings never drift between code and database.
+ """
+
+ __tablename__ = "system_setting"
+
+ id: Mapped[uuid.UUID] = mapped_column(
+ PG_UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+ )
+ # Registry key (e.g. "maintenance_mode"). Unique: at most one override row
+ # per setting. The unique constraint also creates the lookup index.
+ key: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
+ # Serialised value ("true"/"42"/free text). Text, not String(n): the
+ # maintenance message has no natural length cap.
+ value: Mapped[str] = mapped_column(Text, nullable=False)
+ # Admin who last changed it. SET NULL keeps the override readable after the
+ # admin account is deleted.
+ updated_by: Mapped[uuid.UUID | None] = mapped_column(
+ ForeignKey("user.id", ondelete="SET NULL"), default=None
+ )
+ updated_at: Mapped[datetime] = mapped_column(
+ DateTime(timezone=True), default=utc_now, onupdate=utc_now, nullable=False
+ )
+
+ # Admin who last changed the value, surfaced in the admin settings UI.
+ updater: Mapped["User | None"] = relationship("User")
diff --git a/app/repositories/system_setting.py b/app/repositories/system_setting.py
new file mode 100644
index 0000000..bdfe3b8
--- /dev/null
+++ b/app/repositories/system_setting.py
@@ -0,0 +1,54 @@
+import uuid
+from collections.abc import Sequence
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models.system_setting import SystemSetting
+
+
+async def get_system_setting_by_key(
+ session: AsyncSession, key: str
+) -> SystemSetting | None:
+ """Return the override row for ``key``, or ``None`` if no override exists.
+
+ Used on the value-resolution hot path; the editor relationship is not
+ eager-loaded here since callers only read ``value``.
+ """
+ statement = select(SystemSetting).where(SystemSetting.key == key)
+ return (await session.execute(statement)).scalars().first()
+
+
+async def get_all_system_settings(session: AsyncSession) -> Sequence[SystemSetting]:
+ """Return every override row with its last editor loaded (admin view).
+
+ ``updater`` is eager-loaded with ``selectinload`` so serializing the admin
+ settings page never triggers a per-row lazy load (no N+1).
+ """
+ statement = select(SystemSetting).options(selectinload(SystemSetting.updater))
+ return (await session.execute(statement)).scalars().all()
+
+
+async def upsert_system_setting(
+ session: AsyncSession,
+ *,
+ key: str,
+ value: str,
+ updated_by: uuid.UUID | None,
+) -> SystemSetting:
+ """Create or update the override row for ``key`` and return it.
+
+ Inserting applies the ``utc_now`` default to ``updated_at``; updating an
+ existing row bumps it via the column's ``onupdate`` hook.
+ """
+ setting = await get_system_setting_by_key(session, key)
+ if setting is None:
+ setting = SystemSetting(key=key, value=value, updated_by=updated_by)
+ session.add(setting)
+ else:
+ setting.value = value
+ setting.updated_by = updated_by
+ await session.commit()
+ await session.refresh(setting)
+ return setting
diff --git a/app/schemas/admin_permission.py b/app/schemas/admin_permission.py
index c22323e..2fc8fdd 100644
--- a/app/schemas/admin_permission.py
+++ b/app/schemas/admin_permission.py
@@ -26,3 +26,5 @@ class Permission(StrEnum):
STATS_READ = "stats:read"
BROADCAST_READ = "broadcast:read"
BROADCAST_WRITE = "broadcast:write"
+ SYSTEM_SETTINGS_READ = "system_settings:read"
+ SYSTEM_SETTINGS_WRITE = "system_settings:write"
diff --git a/app/schemas/file.py b/app/schemas/file.py
index f0d6bc9..4c4de96 100644
--- a/app/schemas/file.py
+++ b/app/schemas/file.py
@@ -16,6 +16,7 @@ class FileCategory(StrEnum):
GENERAL = "general"
USER_PROFILE_PHOTO = "user_profile_photo"
SUPPORT_ATTACHMENT = "support_attachment"
+ BRANDING_LOGO = "branding_logo"
class FilePublic(BaseModel):
diff --git a/app/schemas/system_settings.py b/app/schemas/system_settings.py
new file mode 100644
index 0000000..8a0dd28
--- /dev/null
+++ b/app/schemas/system_settings.py
@@ -0,0 +1,44 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from app.core.settings_registry import SettingCategory, SettingValue, SettingValueType
+
+
+class SettingRead(BaseModel):
+ """One setting's effective state: registry metadata + resolved value."""
+
+ key: str
+ value: SettingValue
+ value_type: SettingValueType
+ category: SettingCategory
+ is_public: bool
+ description: str
+ updated_at: datetime | None = None
+ updated_by: uuid.UUID | None = None
+
+
+class SettingsListResponse(BaseModel):
+ """Admin view: every setting with metadata and current value."""
+
+ data: list[SettingRead]
+
+
+class PublicSettingsResponse(BaseModel):
+ """Unauthenticated view: only ``is_public`` settings as key -> value."""
+
+ data: dict[str, SettingValue]
+
+
+class SettingUpdate(BaseModel):
+ """Admin request body to change one setting's value."""
+
+ value: SettingValue
+
+
+class SettingUpdateResponse(BaseModel):
+ """The updated setting plus a success message key."""
+
+ setting: SettingRead
+ message: str
diff --git a/app/schemas/user_activity.py b/app/schemas/user_activity.py
index 69e543d..80300ea 100644
--- a/app/schemas/user_activity.py
+++ b/app/schemas/user_activity.py
@@ -29,6 +29,7 @@ class ResourceType(StrEnum):
ACTIVITY = "activity"
SUPPORT_TICKET = "support_ticket"
ANNOUNCEMENT = "announcement"
+ SYSTEM_SETTINGS = "system_settings"
class ActivityStatus(StrEnum):
diff --git a/app/services/auth_service.py b/app/services/auth_service.py
index 91eadb9..574a1f0 100644
--- a/app/services/auth_service.py
+++ b/app/services/auth_service.py
@@ -55,6 +55,7 @@
from app.schemas.user_activity import ActivityStatus, ActivityType, ResourceType
from app.services.user_service import create_user_service
from app.use_cases.log_activity import log_activity
+from app.use_cases.settings import get_bool
from app.utils.email_templates import (
generate_account_locked_email,
generate_email_verification_email,
@@ -72,6 +73,12 @@ async def register_service(
so a self-service registrant can never escalate to admin/superadmin or
self-verify. Privileged user creation goes through the admin flow instead.
"""
+ if not await get_bool(session, "registration_enabled"):
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail=ErrorMessages.REGISTRATION_DISABLED,
+ )
+
safe_user = UserCreate(
email=user_register.email,
password=user_register.password,
diff --git a/app/services/file_service.py b/app/services/file_service.py
index 523c886..af0bc16 100644
--- a/app/services/file_service.py
+++ b/app/services/file_service.py
@@ -10,8 +10,10 @@
from app.models.user import User
from app.repositories.file import create_file
from app.schemas.file import FileCategory
+from app.schemas.user import SystemRole
from app.schemas.user_activity import ActivityType, ResourceType
from app.use_cases.log_activity import log_activity
+from app.use_cases.settings import get_int
logger = logging.getLogger(__name__)
@@ -36,8 +38,23 @@ async def upload_file_service(
and the upload is recorded in the activity log. ``category`` tags the file
and selects the Cloudinary sub-folder it lands in.
"""
+ # The branding logo is a site-wide asset: only admins/superadmins may write
+ # to it, otherwise any active user could overwrite the site logo folder.
+ if category is FileCategory.BRANDING_LOGO and current_user.role not in (
+ SystemRole.ADMIN,
+ SystemRole.SUPERADMIN,
+ ):
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail=ErrorMessages.INSUFFICIENT_PERMISSIONS,
+ )
+
+ # Upload ceiling is admin-configurable at runtime (registry default seeds
+ # from MAX_UPLOAD_SIZE_BYTES, so behaviour is unchanged until an admin edits).
+ max_bytes = await get_int(session, "max_upload_size_mb") * 1024 * 1024
+
# Fast reject using the reported size before reading the body into memory.
- if upload.size is not None and upload.size > settings.MAX_UPLOAD_SIZE_BYTES:
+ if upload.size is not None and upload.size > max_bytes:
raise HTTPException(
status_code=status.HTTP_413_CONTENT_TOO_LARGE,
detail=ErrorMessages.FILE_TOO_LARGE,
@@ -56,15 +73,21 @@ async def upload_file_service(
status_code=status.HTTP_400_BAD_REQUEST,
detail=ErrorMessages.FILE_EMPTY,
)
- if len(content) > settings.MAX_UPLOAD_SIZE_BYTES:
+ if len(content) > max_bytes:
raise HTTPException(
status_code=status.HTTP_413_CONTENT_TOO_LARGE,
detail=ErrorMessages.FILE_TOO_LARGE,
)
# Organise assets on Cloudinary as ``//`` so files
- # are grouped by purpose and owner instead of dumped in one flat folder.
- folder = f"{settings.CLOUDINARY_UPLOAD_FOLDER}/{category.value}/{current_user.id}"
+ # are grouped by purpose and owner. Site-wide assets (e.g. the logo) are not
+ # owned by a user, so they live directly under ``/``.
+ if category is FileCategory.BRANDING_LOGO:
+ folder = f"{settings.CLOUDINARY_UPLOAD_FOLDER}/{category.value}"
+ else:
+ folder = (
+ f"{settings.CLOUDINARY_UPLOAD_FOLDER}/{category.value}/{current_user.id}"
+ )
try:
result = await storage.upload_file(
content, folder=folder, resource_type="image"
diff --git a/app/services/support_service.py b/app/services/support_service.py
index 3cc512b..7b8d0b8 100644
--- a/app/services/support_service.py
+++ b/app/services/support_service.py
@@ -41,6 +41,7 @@
)
from app.schemas.user_activity import ActivityType, ResourceType
from app.use_cases.log_activity import log_activity
+from app.use_cases.settings import get_bool
from app.use_cases.support_attachments import resolve_attachment_files
from app.utils import utc_now
@@ -120,6 +121,12 @@ async def create_ticket_service(
request: Request | None = None,
) -> SupportTicketResponse:
"""Open a new ticket with its first message and optional attachments."""
+ if not await get_bool(session, "support_enabled"):
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail=ErrorMessages.SUPPORT_DISABLED,
+ )
+
files = await resolve_attachment_files(
session,
file_ids=payload.attachment_file_ids,
diff --git a/app/services/system_settings.py b/app/services/system_settings.py
new file mode 100644
index 0000000..020eecd
--- /dev/null
+++ b/app/services/system_settings.py
@@ -0,0 +1,127 @@
+from fastapi import HTTPException, Request, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.messages.error_message import ErrorMessages
+from app.core.messages.success_message import SuccessMessages
+from app.core.settings_registry import (
+ SETTINGS_REGISTRY,
+ SettingDefinition,
+ SettingValue,
+ SettingValueType,
+ coerce,
+ to_str,
+)
+from app.models.system_setting import SystemSetting
+from app.models.user import User
+from app.repositories.system_setting import (
+ get_all_system_settings,
+ upsert_system_setting,
+)
+from app.schemas.system_settings import (
+ PublicSettingsResponse,
+ SettingRead,
+ SettingsListResponse,
+ SettingUpdate,
+ SettingUpdateResponse,
+)
+from app.schemas.user_activity import ActivityType, ResourceType
+from app.use_cases.log_activity import log_activity
+
+
+def _type_matches(value_type: SettingValueType, value: SettingValue) -> bool:
+ """True if ``value``'s Python type matches the declared setting type."""
+ if value_type is SettingValueType.BOOL:
+ return isinstance(value, bool)
+ if value_type is SettingValueType.INT:
+ return isinstance(value, int) and not isinstance(value, bool)
+ return isinstance(value, str)
+
+
+def _to_read(
+ definition: SettingDefinition,
+ value: SettingValue,
+ row: SystemSetting | None,
+) -> SettingRead:
+ """Build the API read model from a definition, its value, and optional row."""
+ return SettingRead(
+ key=definition.key,
+ value=value,
+ value_type=definition.value_type,
+ category=definition.category,
+ is_public=definition.is_public,
+ description=definition.description,
+ updated_at=row.updated_at if row else None,
+ updated_by=row.updated_by if row else None,
+ )
+
+
+async def _resolved_items(session: AsyncSession) -> list[SettingRead]:
+ """Merge the registry with override rows into a full read list (one query)."""
+ rows = {r.key: r for r in await get_all_system_settings(session)}
+ items: list[SettingRead] = []
+ for key, definition in SETTINGS_REGISTRY.items():
+ row = rows.get(key)
+ value = definition.default if row is None else coerce(definition, row.value)
+ items.append(_to_read(definition, value, row))
+ return items
+
+
+async def list_all_settings_service(session: AsyncSession) -> SettingsListResponse:
+ """Return every setting with metadata and value, for the admin UI."""
+ return SettingsListResponse(data=await _resolved_items(session))
+
+
+async def get_public_settings_service(
+ session: AsyncSession,
+) -> PublicSettingsResponse:
+ """Return only public settings as key -> value, for unauthenticated clients."""
+ items = await _resolved_items(session)
+ return PublicSettingsResponse(
+ data={item.key: item.value for item in items if item.is_public}
+ )
+
+
+async def update_setting_service(
+ session: AsyncSession,
+ *,
+ key: str,
+ payload: SettingUpdate,
+ admin: User,
+ request: Request | None = None,
+) -> SettingUpdateResponse:
+ """Validate and persist an admin's change to one setting, then audit it."""
+ definition = SETTINGS_REGISTRY.get(key)
+ if definition is None:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=ErrorMessages.SETTING_NOT_FOUND,
+ )
+ if not _type_matches(definition.value_type, payload.value):
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+ detail=ErrorMessages.SETTING_INVALID_VALUE,
+ )
+ if definition.validate is not None:
+ try:
+ definition.validate(payload.value)
+ except ValueError as exc:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+ detail=ErrorMessages.SETTING_INVALID_VALUE,
+ ) from exc
+
+ row = await upsert_system_setting(
+ session, key=key, value=to_str(payload.value), updated_by=admin.id
+ )
+ await log_activity(
+ session,
+ user_id=admin.id,
+ activity_type=ActivityType.UPDATE,
+ resource_type=ResourceType.SYSTEM_SETTINGS,
+ resource_id=row.id,
+ request=request,
+ )
+ return SettingUpdateResponse(
+ setting=_to_read(definition, coerce(definition, row.value), row),
+ message=SuccessMessages.SETTINGS_UPDATED,
+ )
diff --git a/app/tests/admin/test_system_settings.py b/app/tests/admin/test_system_settings.py
new file mode 100644
index 0000000..7c7f78a
--- /dev/null
+++ b/app/tests/admin/test_system_settings.py
@@ -0,0 +1,155 @@
+"""Tests for the system-settings endpoints and the maintenance-mode middleware.
+
+Covers the admin read/patch endpoints (RBAC, type + registry validation,
+unknown key), the unauthenticated public endpoint (only public keys leak), and
+the maintenance gate (non-admins blocked with 503, admins and exempt paths pass).
+"""
+
+import pytest
+from httpx import AsyncClient
+
+from app.core.messages.error_message import ErrorMessages
+from app.core.messages.success_message import SuccessMessages
+from app.repositories.system_setting import upsert_system_setting
+from app.schemas.admin_permission import Permission
+from app.tests.admin.conftest import (
+ grant_permissions,
+ login,
+ promote_to_admin,
+ register_and_verify,
+)
+from app.tests.conftest import TestingSessionLocal
+
+
+async def _enable_maintenance() -> None:
+ """Turn maintenance mode on directly in the DB (skips the admin endpoint)."""
+ async with TestingSessionLocal() as session:
+ await upsert_system_setting(
+ session, key="maintenance_mode", value="true", updated_by=None
+ )
+
+
+# --- Admin endpoints ---------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_list_settings_returns_registry(admin_client: AsyncClient):
+ """The admin list returns every registered setting with its metadata."""
+ response = await admin_client.get("/admin/system-settings")
+ assert response.status_code == 200, response.text
+ keys = {item["key"] for item in response.json()["data"]}
+ assert {"maintenance_mode", "registration_enabled", "site_name"} <= keys
+
+
+@pytest.mark.asyncio
+async def test_patch_setting_updates_value(admin_client: AsyncClient):
+ """A valid PATCH persists the new value and returns the success message."""
+ response = await admin_client.patch(
+ "/admin/system-settings/maintenance_mode", json={"value": True}
+ )
+ assert response.status_code == 200, response.text
+ body = response.json()
+ assert body["setting"]["value"] is True
+ assert body["message"] == SuccessMessages.SETTINGS_UPDATED
+
+
+@pytest.mark.asyncio
+async def test_patch_unknown_key_returns_404(admin_client: AsyncClient):
+ """An unknown setting key is rejected with 404."""
+ response = await admin_client.patch(
+ "/admin/system-settings/does_not_exist", json={"value": True}
+ )
+ assert response.status_code == 404
+ assert response.json()["error"] == ErrorMessages.SETTING_NOT_FOUND
+
+
+@pytest.mark.asyncio
+async def test_patch_wrong_type_returns_422(admin_client: AsyncClient):
+ """A value whose type mismatches the setting is rejected with 422."""
+ response = await admin_client.patch(
+ "/admin/system-settings/maintenance_mode", json={"value": "not-a-bool"}
+ )
+ assert response.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_patch_invalid_value_returns_422(admin_client: AsyncClient):
+ """A value failing the registry validator is rejected with 422."""
+ response = await admin_client.patch(
+ "/admin/system-settings/default_locale", json={"value": "de"}
+ )
+ assert response.status_code == 422
+ assert response.json()["error"] == ErrorMessages.SETTING_INVALID_VALUE
+
+
+@pytest.mark.asyncio
+async def test_settings_require_permission(client: AsyncClient):
+ """An admin lacking system_settings perms is forbidden on read and write."""
+ await register_and_verify(client, "limited@test.com")
+ await promote_to_admin("limited@test.com")
+ await grant_permissions("limited@test.com", [Permission.USERS_READ])
+ await login(client, "limited@test.com")
+
+ assert (await client.get("/admin/system-settings")).status_code == 403
+ patched = await client.patch(
+ "/admin/system-settings/maintenance_mode", json={"value": True}
+ )
+ assert patched.status_code == 403
+
+
+# --- Public endpoint ---------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_public_settings_exposes_only_public(client: AsyncClient):
+ """The unauthenticated endpoint returns public keys and hides private ones."""
+ response = await client.get("/settings/public")
+ assert response.status_code == 200, response.text
+ data = response.json()["data"]
+ assert "site_name" in data
+ assert "maintenance_mode" in data
+ # max_upload_size_mb is is_public=False — it must never leak here.
+ assert "max_upload_size_mb" not in data
+
+
+# --- Maintenance-mode middleware ---------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_maintenance_blocks_anonymous(client: AsyncClient):
+ """With maintenance on, an anonymous request to a gated path gets 503."""
+ await _enable_maintenance()
+ response = await client.get("/users/me")
+ assert response.status_code == 503
+ assert response.json()["error"] == ErrorMessages.MAINTENANCE_MODE
+
+
+@pytest.mark.asyncio
+async def test_maintenance_blocks_regular_user(regular_client: AsyncClient):
+ """With maintenance on, a non-admin user is blocked with 503."""
+ await _enable_maintenance()
+ assert (await regular_client.get("/users/me")).status_code == 503
+
+
+@pytest.mark.asyncio
+async def test_maintenance_allows_admin(admin_client: AsyncClient):
+ """Admins keep working while maintenance mode is on."""
+ await _enable_maintenance()
+ assert (await admin_client.get("/users/me")).status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_maintenance_exempts_public_settings(client: AsyncClient):
+ """The public settings endpoint stays reachable during maintenance."""
+ await _enable_maintenance()
+ assert (await client.get("/settings/public")).status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_maintenance_exempts_login(client: AsyncClient):
+ """Login stays reachable during maintenance so admins can authenticate."""
+ await _enable_maintenance()
+ response = await client.post(
+ "/auth/login", data={"username": "nobody@test.com", "password": "x"}
+ )
+ assert response.status_code != 503
diff --git a/app/tests/test_files.py b/app/tests/test_files.py
index 9eed674..89825fd 100644
--- a/app/tests/test_files.py
+++ b/app/tests/test_files.py
@@ -7,6 +7,7 @@
from app.core.config import settings
from app.models.file import File
from app.models.user import User
+from app.repositories.system_setting import upsert_system_setting
from app.tests.conftest import TestingSessionLocal
PNG_BYTES = b"\x89PNG\r\n\x1a\nfake-image-bytes"
@@ -48,6 +49,15 @@ async def _upload(
return await client.post("/upload", files={"file": (name, data, content_type)})
+async def _promote_to_admin(email: str) -> None:
+ """Grant the user the admin role directly in the DB."""
+ async with TestingSessionLocal() as session:
+ await session.execute(
+ update(User).where(User.email == email).values(role="admin")
+ )
+ await session.commit()
+
+
@pytest.mark.asyncio
async def test_upload_requires_auth(client: AsyncClient):
"""Anonymous callers cannot upload."""
@@ -104,6 +114,42 @@ async def test_upload_rejects_unknown_category(client: AsyncClient):
assert response.status_code == 422
+@pytest.mark.asyncio
+async def test_upload_branding_logo_forbidden_for_regular_user(client: AsyncClient):
+ """A non-admin cannot write to the site-wide branding logo bucket."""
+ await _register_verify_login(client, "u@test.com")
+
+ response = await client.post(
+ "/upload",
+ files={"file": ("logo.png", PNG_BYTES, "image/png")},
+ data={"category": "branding_logo"},
+ )
+
+ assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_upload_branding_logo_allowed_for_admin(
+ client: AsyncClient, mock_cloudinary
+):
+ """An admin may upload the branding logo; it lands in the site-wide folder."""
+ await _register_verify_login(client, "admin@test.com")
+ await _promote_to_admin("admin@test.com")
+
+ response = await client.post(
+ "/upload",
+ files={"file": ("logo.png", PNG_BYTES, "image/png")},
+ data={"category": "branding_logo"},
+ )
+
+ assert response.status_code == 201, response.text
+ assert response.json()["category"] == "branding_logo"
+
+ # Site-wide asset: no per-user sub-folder, unlike user-owned categories.
+ folder = mock_cloudinary.upload.call_args.kwargs["folder"]
+ assert folder == f"{settings.CLOUDINARY_UPLOAD_FOLDER}/branding_logo"
+
+
@pytest.mark.asyncio
async def test_upload_invalid_mime_rejected(client: AsyncClient):
"""Non-image content types are rejected with 415."""
@@ -121,11 +167,16 @@ async def test_upload_empty_rejected(client: AsyncClient):
@pytest.mark.asyncio
-async def test_upload_too_large_rejected(client: AsyncClient, monkeypatch):
- """Files over MAX_UPLOAD_SIZE_BYTES are rejected with 413."""
+async def test_upload_too_large_rejected(client: AsyncClient):
+ """Files over the configured max upload size are rejected with 413."""
await _register_verify_login(client, "u@test.com")
- monkeypatch.setattr(settings, "MAX_UPLOAD_SIZE_BYTES", 4)
- response = await _upload(client, data=b"too-many-bytes")
+ # Lower the limit to its 1 MB minimum via the system setting, then exceed it.
+ async with TestingSessionLocal() as session:
+ await upsert_system_setting(
+ session, key="max_upload_size_mb", value="1", updated_by=None
+ )
+ oversized = b"0" * (1024 * 1024 + 1)
+ response = await _upload(client, data=oversized)
assert response.status_code == 413
diff --git a/app/use_cases/settings.py b/app/use_cases/settings.py
new file mode 100644
index 0000000..5a44115
--- /dev/null
+++ b/app/use_cases/settings.py
@@ -0,0 +1,54 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.settings_registry import (
+ SETTINGS_REGISTRY,
+ SettingDefinition,
+ SettingValue,
+ SettingValueType,
+ coerce,
+)
+from app.repositories.system_setting import get_system_setting_by_key
+
+
+async def _resolve(
+ session: AsyncSession, key: str
+) -> tuple[SettingDefinition, SettingValue]:
+ """Return a setting's definition and effective value (DB override or default)."""
+ definition = SETTINGS_REGISTRY.get(key)
+ if definition is None:
+ raise KeyError(f"unknown setting key: {key}")
+ row = await get_system_setting_by_key(session, key)
+ value = definition.default if row is None else coerce(definition, row.value)
+ return definition, value
+
+
+async def get_bool(session: AsyncSession, key: str) -> bool:
+ """Resolve a boolean setting; raises if the key is not a bool setting."""
+ definition, value = await _resolve(session, key)
+ if definition.value_type is not SettingValueType.BOOL or not isinstance(
+ value, bool
+ ):
+ raise TypeError(f"{key} is not a boolean setting")
+ return value
+
+
+async def get_int(session: AsyncSession, key: str) -> int:
+ """Resolve an integer setting; raises if the key is not an int setting."""
+ definition, value = await _resolve(session, key)
+ if (
+ definition.value_type is not SettingValueType.INT
+ or not isinstance(value, int)
+ or isinstance(value, bool)
+ ):
+ raise TypeError(f"{key} is not an integer setting")
+ return value
+
+
+async def get_str(session: AsyncSession, key: str) -> str:
+ """Resolve a string setting; raises if the key is not a string setting."""
+ definition, value = await _resolve(session, key)
+ if definition.value_type is not SettingValueType.STRING or not isinstance(
+ value, str
+ ):
+ raise TypeError(f"{key} is not a string setting")
+ return value