diff --git a/app/cabinet/routes/admin_news.py b/app/cabinet/routes/admin_news.py
index c201ec75..5eac7747 100644
--- a/app/cabinet/routes/admin_news.py
+++ b/app/cabinet/routes/admin_news.py
@@ -1,6 +1,8 @@
 """Admin routes for managing news articles in cabinet."""
 
+import asyncio
 from datetime import UTC, datetime
+from typing import Any
 
 import structlog
 from fastapi import APIRouter, Depends, HTTPException, Query, status
@@ -13,10 +15,9 @@ from app.database.crud.news import (
     get_all_news,
     get_all_news_count,
     get_news_article_by_id,
-    get_news_article_by_slug,
     update_news_article,
 )
-from app.database.models import User
+from app.database.models import NewsArticle, User
 
 from ..dependencies import get_cabinet_db, require_permission
 from ..schemas.news import (
@@ -34,9 +35,12 @@ logger = structlog.get_logger(__name__)
 router = APIRouter(prefix='/admin/news', tags=['Cabinet Admin News'])
 
 
-def _article_to_detail(article) -> dict:
-    """Convert NewsArticle ORM instance to full detail dict."""
-    author_name = None
+def _article_to_detail(article: NewsArticle) -> dict[str, Any]:
+    """Convert NewsArticle ORM instance to full detail dict.
+
+    Expects the ``author`` relationship to be eagerly loaded.
+    """
+    author_name: str | None = None
     if article.author:
         author_name = article.author.first_name or article.author.username or f'#{article.author.id}'
 
@@ -68,35 +72,23 @@ async def list_all_news(
     limit: int = Query(50, ge=1, le=200),
     offset: int = Query(0, ge=0),
 ) -> NewsListResponse:
-    """Get all news articles (admin view, includes unpublished)."""
-    try:
-        articles = await get_all_news(db, limit=limit, offset=offset)
-        total = await get_all_news_count(db)
+    """Get all news articles (admin view, includes unpublished).
 
-        items = [
-            NewsArticleListItem(
-                id=a.id,
-                title=a.title,
-                slug=a.slug,
-                excerpt=a.excerpt,
-                category=a.category,
-                category_color=a.category_color,
-                tag=a.tag,
-                featured_image_url=a.featured_image_url,
-                is_published=a.is_published,
-                is_featured=a.is_featured,
-                published_at=a.published_at,
-                read_time_minutes=a.read_time_minutes,
-                views_count=a.views_count,
-            )
-            for a in articles
-        ]
+    articles and total are independent — run them concurrently.
+    """
+    try:
+        articles, total = await asyncio.gather(
+            get_all_news(db, limit=limit, offset=offset),
+            get_all_news_count(db),
+        )
+
+        items = [NewsArticleListItem.model_validate(a) for a in articles]
 
         return NewsListResponse(items=items, total=total)
     except HTTPException:
         raise
-    except Exception as e:
-        logger.error('Failed to list all news', error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to list all news')
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to load news articles',
@@ -127,14 +119,6 @@ async def create_article(
     db: AsyncSession = Depends(get_cabinet_db),
 ) -> NewsArticleResponse:
     """Create a new news article."""
-    # Check slug uniqueness
-    existing = await get_news_article_by_slug(db, request.slug)
-    if existing:
-        raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
-            detail='An article with this slug already exists',
-        )
-
     try:
         article = await create_news_article(
             db,
@@ -148,30 +132,30 @@ async def create_article(
             featured_image_url=request.featured_image_url,
             is_published=request.is_published,
             is_featured=request.is_featured,
-            published_at=None,
             read_time_minutes=request.read_time_minutes,
             created_by=admin.id,
         )
-
-        # Reload with author relationship
-        article = await get_news_article_by_id(db, article.id)
-        if not article:
-            raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail='Failed to reload article')
-        return NewsArticleResponse(**_article_to_detail(article))
-    except HTTPException:
-        raise
     except IntegrityError:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
             detail='An article with this slug already exists',
         )
-    except Exception as e:
-        logger.error('Failed to create news article', error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to create news article')
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to create article',
         )
 
+    # Reload with author relationship
+    article = await get_news_article_by_id(db, article.id)
+    if not article:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail='Failed to reload article after creation',
+        )
+    return NewsArticleResponse(**_article_to_detail(article))
+
 
 @router.put('/{article_id}', response_model=NewsArticleResponse)
 async def update_article(
@@ -188,38 +172,30 @@ async def update_article(
             detail='Article not found',
         )
 
-    # Check slug uniqueness if slug is being changed
-    if request.slug and request.slug != article.slug:
-        existing = await get_news_article_by_slug(db, request.slug)
-        if existing:
-            raise HTTPException(
-                status_code=status.HTTP_409_CONFLICT,
-                detail='An article with this slug already exists',
-            )
-
     try:
         update_data = request.model_dump(exclude_unset=True)
         article = await update_news_article(db, article, **update_data)
-
-        # Reload with author relationship
-        article = await get_news_article_by_id(db, article.id)
-        if not article:
-            raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail='Failed to reload article')
-        return NewsArticleResponse(**_article_to_detail(article))
-    except HTTPException:
-        raise
     except IntegrityError:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
             detail='An article with this slug already exists',
         )
-    except Exception as e:
-        logger.error('Failed to update news article', article_id=article_id, error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to update news article', article_id=article_id)
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to update article',
         )
 
+    # Reload with author relationship (update used bulk UPDATE, author not populated)
+    article = await get_news_article_by_id(db, article.id)
+    if not article:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail='Failed to reload article after update',
+        )
+    return NewsArticleResponse(**_article_to_detail(article))
+
 
 @router.delete('/{article_id}', status_code=status.HTTP_204_NO_CONTENT)
 async def remove_article(
@@ -237,8 +213,8 @@ async def remove_article(
 
     try:
         await delete_news_article(db, article)
-    except Exception as e:
-        logger.error('Failed to delete news article', article_id=article_id, error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to delete news article', article_id=article_id)
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to delete article',
@@ -261,7 +237,7 @@ async def toggle_publish(
 
     new_published = not article.is_published
 
-    update_kwargs: dict = {'is_published': new_published}
+    update_kwargs: dict[str, Any] = {'is_published': new_published}
     # Auto-set published_at on first publish
     if new_published and article.published_at is None:
         update_kwargs['published_at'] = datetime.now(UTC)
@@ -274,8 +250,8 @@ async def toggle_publish(
             is_featured=article.is_featured,
             published_at=article.published_at,
         )
-    except Exception as e:
-        logger.error('Failed to toggle publish', article_id=article_id, error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to toggle publish', article_id=article_id)
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to toggle publish status',
@@ -304,8 +280,8 @@ async def toggle_featured(
             is_featured=article.is_featured,
             published_at=article.published_at,
         )
-    except Exception as e:
-        logger.error('Failed to toggle featured', article_id=article_id, error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to toggle featured', article_id=article_id)
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to toggle featured status',
diff --git a/app/cabinet/routes/news.py b/app/cabinet/routes/news.py
index 4bf6f11c..6b3cc2bf 100644
--- a/app/cabinet/routes/news.py
+++ b/app/cabinet/routes/news.py
@@ -1,7 +1,11 @@
 """Public news routes for cabinet - user-facing news/blog section."""
 
+import asyncio
+import time
+from typing import Any
+
 import structlog
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.database.crud.news import (
@@ -11,7 +15,7 @@ from app.database.crud.news import (
     get_published_news_count,
     increment_views,
 )
-from app.database.models import User
+from app.database.models import NewsArticle, User
 
 from ..dependencies import get_cabinet_db, get_current_cabinet_user
 from ..schemas.news import (
@@ -23,16 +27,51 @@ from ..schemas.news import (
 
 logger = structlog.get_logger(__name__)
 
+# Slug constraint: alphanumeric, hyphens, underscores, max 500 chars
+_SLUG_MAX_LENGTH: int = 500
+_SLUG_PATTERN: str = r'^[a-zA-Z0-9_-]+$'
+
+# --- View counter deduplication ---
+# In-memory TTL cache to prevent a single user from inflating view counts.
+# Key: (user_id, article_id), Value: timestamp of last counted view.
+# Views from the same user on the same article within _VIEW_DEDUP_SECONDS are ignored.
+_VIEW_DEDUP_SECONDS: int = 300  # 5 minutes
+_VIEW_DEDUP_MAX_SIZE: int = 10_000  # max entries before eviction
+_view_dedup_cache: dict[tuple[int, int], float] = {}
+
+
+def _should_count_view(user_id: int, article_id: int) -> bool:
+    """Return True if this view should be counted (not a duplicate within TTL)."""
+    now = time.monotonic()
+    key = (user_id, article_id)
+    last_seen = _view_dedup_cache.get(key)
+
+    if last_seen is not None and (now - last_seen) < _VIEW_DEDUP_SECONDS:
+        return False
+
+    # Evict stale entries if cache grows too large
+    if len(_view_dedup_cache) >= _VIEW_DEDUP_MAX_SIZE:
+        cutoff = now - _VIEW_DEDUP_SECONDS
+        stale_keys = [k for k, v in _view_dedup_cache.items() if v < cutoff]
+        for k in stale_keys:
+            del _view_dedup_cache[k]
+
+    _view_dedup_cache[key] = now
+    return True
+
+
 router = APIRouter(prefix='/news', tags=['Cabinet News'])
 
 
-def _article_to_response(article, *, include_content: bool = True) -> dict:
-    """Convert NewsArticle ORM instance to response dict."""
-    author_name = None
-    if article.author:
-        author_name = article.author.first_name or article.author.username or f'#{article.author.id}'
+def _article_to_response(article: NewsArticle, *, include_content: bool = True) -> dict[str, Any]:
+    """Convert NewsArticle ORM instance to response dict.
 
-    data = {
+    ``author_name`` is only resolved when ``include_content=True`` (single-article
+    detail view) because the author relationship is not eagerly loaded for list
+    queries -- accessing it there would trigger a lazy-load or raise
+    ``MissingGreenlet`` in async context.
+    """
+    data: dict[str, Any] = {
         'id': article.id,
         'title': article.title,
         'slug': article.slug,
@@ -49,6 +88,9 @@ def _article_to_response(article, *, include_content: bool = True) -> dict:
     }
 
     if include_content:
+        author_name: str | None = None
+        if article.author:
+            author_name = article.author.first_name or article.author.username or f'#{article.author.id}'
         data['content'] = article.content
         data['author_name'] = author_name
         data['created_at'] = article.created_at
@@ -66,8 +108,8 @@ async def list_categories(
     """Get list of distinct news categories."""
     try:
         return await get_news_categories(db)
-    except Exception as e:
-        logger.error('Failed to get news categories', error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to get news categories')
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to load categories',
@@ -82,19 +124,26 @@ async def list_published_news(
     limit: int = Query(20, ge=1, le=100),
     offset: int = Query(0, ge=0),
 ) -> NewsListResponse:
-    """Get paginated list of published news articles."""
+    """Get paginated list of published news articles.
+
+    The three DB queries (articles, count, categories) are independent — run
+    them concurrently via asyncio.gather to cut latency to the slowest query
+    instead of the sequential sum of all three.
+    """
     try:
-        articles = await get_published_news(db, category=category, limit=limit, offset=offset)
-        total = await get_published_news_count(db, category=category)
-        categories = await get_news_categories(db)
+        articles, total, categories = await asyncio.gather(
+            get_published_news(db, category=category, limit=limit, offset=offset),
+            get_published_news_count(db, category=category),
+            get_news_categories(db),
+        )
 
         items = [NewsArticleListItem(**_article_to_response(a, include_content=False)) for a in articles]
 
         return NewsListResponse(items=items, total=total, categories=categories)
     except HTTPException:
         raise
-    except Exception as e:
-        logger.error('Failed to list published news', error=str(e), exc_info=True)
+    except Exception:
+        logger.exception('Failed to list published news')
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail='Failed to load news',
@@ -103,7 +152,7 @@ async def list_published_news(
 
 @router.get('/{slug}', response_model=NewsArticleResponse)
 async def get_article_by_slug(
-    slug: str,
+    slug: str = Path(..., max_length=_SLUG_MAX_LENGTH, pattern=_SLUG_PATTERN),
     user: User = Depends(get_current_cabinet_user),
     db: AsyncSession = Depends(get_cabinet_db),
 ) -> NewsArticleResponse:
@@ -116,11 +165,15 @@ async def get_article_by_slug(
             detail='Article not found',
         )
 
-    # Increment views in background-safe manner (no error propagation)
-    try:
-        await increment_views(db, article.id)
-        await db.refresh(article)
-    except Exception:
-        logger.warning('Failed to increment views', article_id=article.id)
+    # Increment views with per-user deduplication (5-min TTL).
+    # Prevents view count inflation from repeated requests by the same user.
+    if _should_count_view(user.id, article.id):
+        try:
+            new_count = await increment_views(db, article.id)
+            # Patch the ORM instance so the response reflects the new count
+            # without an extra db.refresh() round-trip
+            article.views_count = new_count
+        except Exception:
+            logger.warning('Failed to increment views', article_id=article.id)
 
     return NewsArticleResponse(**_article_to_response(article, include_content=True))
diff --git a/app/cabinet/schemas/news.py b/app/cabinet/schemas/news.py
index 2ef6d0e3..3d84ef52 100644
--- a/app/cabinet/schemas/news.py
+++ b/app/cabinet/schemas/news.py
@@ -1,11 +1,35 @@
-"""Schemas for news articles in cabinet."""
+"""Schemas for news articles in cabinet.
+
+Security notes:
+- featured_image_url is validated to only accept http/https schemes.
+- category_color is validated as a strict hex color (#RGB, #RRGGBB, etc.).
+- Slug is sanitized to only allow [a-zA-Z0-9_-].
+- Content is server-side sanitized to strip <script>, event handlers, and
+  dangerous URI schemes as a defense-in-depth measure. The frontend also
+  sanitizes via DOMPurify, but server-side sanitization protects against
+  alternative consumers (mobile apps, RSS, email digests) and compromised
+  frontends.
+"""
 
 import re
 from datetime import datetime
+from urllib.parse import urlparse
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 
+# Pre-compiled regex for hex color validation (reused across validators)
+_HEX_COLOR_RE: re.Pattern[str] = re.compile(r'^#([0-9a-fA-F]{3,4}|[0-9a-fA-F]{6}|[0-9a-fA-F]{8})$')
+
+# Pre-compiled regex for collapsing repeated hyphens in slugs
+_MULTI_HYPHEN_RE: re.Pattern[str] = re.compile(r'-+')
+
+# Maximum slug length (matches DB column constraint)
+_MAX_SLUG_LENGTH: int = 500
+
+# Allowed URL schemes for user-supplied URLs (featured_image_url)
+_SAFE_URL_SCHEMES: frozenset[str] = frozenset({'http', 'https'})
+
 # Cyrillic-to-Latin transliteration map for slug generation
 _TRANSLIT_MAP: dict[str, str] = {
     'а': 'a',
@@ -56,8 +80,88 @@ def _slugify(title: str) -> str:
         elif ch == ' ':
             result.append('-')
     slug = ''.join(result)
-    slug = re.sub(r'-+', '-', slug).strip('-')
-    return slug or 'untitled'
+    slug = _MULTI_HYPHEN_RE.sub('-', slug).strip('-')
+    return slug[:_MAX_SLUG_LENGTH] or 'untitled'
+
+
+def _validate_hex_color(v: str) -> str:
+    """Validate a hex color string. Raises ValueError on invalid input."""
+    if not _HEX_COLOR_RE.match(v):
+        msg = 'category_color must be a valid hex color (e.g. #00e5a0)'
+        raise ValueError(msg)
+    return v
+
+
+def _validate_safe_url(v: str) -> str:
+    """Validate that a URL uses http or https scheme only.
+
+    Prevents javascript:, data:, vbscript:, and other dangerous URI schemes
+    from being stored in the database and later rendered in <img> or <a> tags.
+    """
+    try:
+        parsed = urlparse(v)
+    except Exception:
+        msg = 'Invalid URL format'
+        raise ValueError(msg)
+
+    if parsed.scheme not in _SAFE_URL_SCHEMES:
+        msg = f'URL scheme must be http or https, got: {parsed.scheme!r}'
+        raise ValueError(msg)
+
+    if not parsed.netloc:
+        msg = 'URL must have a valid host'
+        raise ValueError(msg)
+
+    return v
+
+
+# --- Server-side HTML content sanitization ---
+# Pre-compiled patterns for stripping the most dangerous HTML constructs.
+# This is a defense-in-depth measure: the frontend also sanitizes via DOMPurify.
+# Uses regex rather than a full HTML parser to avoid adding a new dependency.
+# Strips: <script>, <style>, <object>, <embed>, <applet>, <base>, <form>,
+# <link>, <meta> tags and all on* event handler attributes.
+_DANGEROUS_TAGS_RE: re.Pattern[str] = re.compile(
+    r'<\s*/?\s*(script|style|object|embed|applet|base|form|link(?:\s)|meta)\b[^>]*>',
+    re.IGNORECASE | re.DOTALL,
+)
+# Match on* event handler attributes, e.g. onclick="...", onerror='...'
+_EVENT_HANDLER_RE: re.Pattern[str] = re.compile(
+    r'\s+on[a-z]+\s*=\s*(?:"[^"]*"|\'[^\']*\'|[^\s>]+)',
+    re.IGNORECASE,
+)
+# Match javascript:, vbscript:, data: in href/src attributes
+_DANGEROUS_URI_RE: re.Pattern[str] = re.compile(
+    r'((?:href|src)\s*=\s*["\'])\s*(javascript|vbscript|data)\s*:',
+    re.IGNORECASE,
+)
+
+
+def _sanitize_html_content(html: str) -> str:
+    """Strip dangerous HTML constructs from article content.
+
+    This is NOT a replacement for DOMPurify on the frontend. It is a
+    defense-in-depth layer that removes the most obvious XSS vectors
+    at the storage boundary. A full HTML sanitizer (nh3, bleach) would
+    be stronger, but this avoids adding a new dependency.
+    """
+    if not html:
+        return html
+
+    # 1. Remove dangerous tags and their content
+    result = _DANGEROUS_TAGS_RE.sub('', html)
+
+    # Also strip <script>...</script> content (tag + body)
+    result = re.sub(r'<script\b[^>]*>[\s\S]*?</script>', '', result, flags=re.IGNORECASE)
+    result = re.sub(r'<style\b[^>]*>[\s\S]*?</style>', '', result, flags=re.IGNORECASE)
+
+    # 2. Remove event handler attributes
+    result = _EVENT_HANDLER_RE.sub('', result)
+
+    # 3. Neutralize dangerous URI schemes in href/src
+    result = _DANGEROUS_URI_RE.sub(r'\1about:', result)
+
+    return result
 
 
 class NewsArticleResponse(BaseModel):
@@ -116,7 +220,7 @@ class NewsCreateRequest(BaseModel):
     """Request to create a news article."""
 
     title: str = Field(..., min_length=1, max_length=500)
-    slug: str | None = Field(None, max_length=500)
+    slug: str | None = Field(None, min_length=1, max_length=500)
     content: str = Field(default='', max_length=500_000)
     excerpt: str | None = Field(None, max_length=1000)
     category: str = Field(..., min_length=1, max_length=100)
@@ -127,27 +231,51 @@ class NewsCreateRequest(BaseModel):
     is_featured: bool = False
     read_time_minutes: int = Field(default=1, ge=1, le=60)
 
+    @field_validator('content')
+    @classmethod
+    def sanitize_content(cls, v: str) -> str:
+        """Strip dangerous HTML from article content (defense-in-depth)."""
+        return _sanitize_html_content(v)
+
     @field_validator('category_color')
     @classmethod
     def validate_hex_color(cls, v: str) -> str:
-        if not re.match(r'^#([0-9a-fA-F]{3,4}|[0-9a-fA-F]{6}|[0-9a-fA-F]{8})$', v):
-            raise ValueError('category_color must be a valid hex color (e.g. #00e5a0)')
+        return _validate_hex_color(v)
+
+    @field_validator('featured_image_url')
+    @classmethod
+    def validate_featured_image_url(cls, v: str | None) -> str | None:
+        """Reject javascript:, data:, and other dangerous URL schemes."""
+        if v is not None:
+            return _validate_safe_url(v)
         return v
 
-    @field_validator('slug', mode='before')
+    @model_validator(mode='before')
     @classmethod
-    def generate_slug(cls, v: str | None, info) -> str:
-        if v:
-            return v
-        title = info.data.get('title', '')
-        return _slugify(title)
+    def auto_generate_slug(cls, data: dict) -> dict:  # type: ignore[type-arg]
+        """Generate slug from title when not explicitly provided."""
+        if isinstance(data, dict) and not data.get('slug'):
+            title = data.get('title', '')
+            data['slug'] = _slugify(title) if isinstance(title, str) else 'untitled'
+        return data
+
+    @field_validator('slug')
+    @classmethod
+    def sanitize_slug(cls, v: str | None) -> str | None:
+        """Ensure slug contains only URL-safe characters."""
+        if v is not None:
+            # Strip anything that isn't alphanumeric, hyphen, or underscore
+            sanitized = re.sub(r'[^a-zA-Z0-9_-]', '-', v)
+            sanitized = _MULTI_HYPHEN_RE.sub('-', sanitized).strip('-')
+            return sanitized or 'untitled'
+        return v
 
 
 class NewsUpdateRequest(BaseModel):
     """Request to update a news article."""
 
     title: str | None = Field(None, min_length=1, max_length=500)
-    slug: str | None = Field(None, max_length=500)
+    slug: str | None = Field(None, min_length=1, max_length=500)
     content: str | None = Field(None, max_length=500_000)
     excerpt: str | None = None
     category: str | None = Field(None, min_length=1, max_length=100)
@@ -158,11 +286,37 @@ class NewsUpdateRequest(BaseModel):
     is_featured: bool | None = None
     read_time_minutes: int | None = Field(None, ge=1, le=60)
 
+    @field_validator('content')
+    @classmethod
+    def sanitize_content(cls, v: str | None) -> str | None:
+        """Strip dangerous HTML from article content (defense-in-depth)."""
+        if v is not None:
+            return _sanitize_html_content(v)
+        return v
+
     @field_validator('category_color')
     @classmethod
     def validate_hex_color(cls, v: str | None) -> str | None:
-        if v is not None and not re.match(r'^#([0-9a-fA-F]{3,4}|[0-9a-fA-F]{6}|[0-9a-fA-F]{8})$', v):
-            raise ValueError('category_color must be a valid hex color (e.g. #00e5a0)')
+        if v is not None:
+            return _validate_hex_color(v)
+        return v
+
+    @field_validator('featured_image_url')
+    @classmethod
+    def validate_featured_image_url(cls, v: str | None) -> str | None:
+        """Reject javascript:, data:, and other dangerous URL schemes."""
+        if v is not None:
+            return _validate_safe_url(v)
+        return v
+
+    @field_validator('slug')
+    @classmethod
+    def sanitize_slug(cls, v: str | None) -> str | None:
+        """Ensure slug contains only URL-safe characters."""
+        if v is not None:
+            sanitized = re.sub(r'[^a-zA-Z0-9_-]', '-', v)
+            sanitized = _MULTI_HYPHEN_RE.sub('-', sanitized).strip('-')
+            return sanitized or 'untitled'
         return v
 
 
diff --git a/app/database/crud/news.py b/app/database/crud/news.py
index 466f9a62..979fbc10 100644
--- a/app/database/crud/news.py
+++ b/app/database/crud/news.py
@@ -1,9 +1,11 @@
 """CRUD operations for news articles."""
 
 from datetime import UTC, datetime
+from typing import Any
 
 import structlog
-from sqlalchemy import delete, func, select, update
+from sqlalchemy import delete, func, nullslast, select, update
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -12,6 +14,34 @@ from app.database.models import NewsArticle
 
 logger = structlog.get_logger(__name__)
 
+# Fields that can be set via update_news_article
+_ALLOWED_UPDATE_FIELDS: frozenset[str] = frozenset(
+    {
+        'title',
+        'slug',
+        'content',
+        'excerpt',
+        'category',
+        'category_color',
+        'tag',
+        'featured_image_url',
+        'is_published',
+        'is_featured',
+        'published_at',
+        'read_time_minutes',
+    }
+)
+
+# Fields that can be explicitly set to None
+_NULLABLE_UPDATE_FIELDS: frozenset[str] = frozenset(
+    {
+        'excerpt',
+        'tag',
+        'featured_image_url',
+        'published_at',
+    }
+)
+
 
 async def create_news_article(
     db: AsyncSession,
@@ -30,7 +60,11 @@ async def create_news_article(
     read_time_minutes: int = 1,
     created_by: int | None = None,
 ) -> NewsArticle:
-    """Create a new news article."""
+    """Create a new news article.
+
+    Raises:
+        IntegrityError: if slug is not unique (caller must handle).
+    """
     # Auto-set published_at when publishing without explicit date
     if is_published and published_at is None:
         published_at = datetime.now(UTC)
@@ -52,7 +86,11 @@ async def create_news_article(
     )
 
     db.add(article)
-    await db.commit()
+    try:
+        await db.commit()
+    except IntegrityError:
+        await db.rollback()
+        raise
     await db.refresh(article)
 
     logger.info(
@@ -87,18 +125,17 @@ async def get_published_news(
     limit: int = 20,
     offset: int = 0,
 ) -> list[NewsArticle]:
-    """Get published news articles, ordered by published_at descending."""
-    stmt = (
-        select(NewsArticle)
-        .options(selectinload(NewsArticle.author))
-        .where(NewsArticle.is_published.is_(True))
-        .order_by(NewsArticle.published_at.desc())
-        .offset(offset)
-        .limit(limit)
-    )
+    """Get published news articles, ordered by published_at descending.
+
+    Does NOT load the author relationship -- list views do not need it.
+    """
+    stmt = select(NewsArticle).where(NewsArticle.is_published.is_(True))
     if category:
         stmt = stmt.where(NewsArticle.category == category)
 
+    # NULLs last so articles without published_at don't float to the top in DESC
+    stmt = stmt.order_by(nullslast(NewsArticle.published_at.desc())).offset(offset).limit(limit)
+
     result = await db.execute(stmt)
     return list(result.scalars().all())
 
@@ -124,13 +161,7 @@ async def get_all_news(
     offset: int = 0,
 ) -> list[NewsArticle]:
     """Get all news articles (admin), ordered by created_at descending."""
-    stmt = (
-        select(NewsArticle)
-        .options(selectinload(NewsArticle.author))
-        .order_by(NewsArticle.created_at.desc())
-        .offset(offset)
-        .limit(limit)
-    )
+    stmt = select(NewsArticle).order_by(NewsArticle.created_at.desc()).offset(offset).limit(limit)
     result = await db.execute(stmt)
     return list(result.scalars().all())
 
@@ -156,36 +187,18 @@ async def get_news_categories(db: AsyncSession) -> list[str]:
 async def update_news_article(
     db: AsyncSession,
     article: NewsArticle,
-    **kwargs,
+    **kwargs: Any,
 ) -> NewsArticle:
-    """Update a news article. Only whitelisted fields are applied."""
-    allowed_fields = {
-        'title',
-        'slug',
-        'content',
-        'excerpt',
-        'category',
-        'category_color',
-        'tag',
-        'featured_image_url',
-        'is_published',
-        'is_featured',
-        'published_at',
-        'read_time_minutes',
-    }
+    """Update a news article. Only whitelisted fields are applied.
 
-    nullable_fields = {
-        'excerpt',
-        'tag',
-        'featured_image_url',
-        'published_at',
-    }
-
-    update_data: dict = {}
+    Raises:
+        IntegrityError: if slug conflicts with another article (caller must handle).
+    """
+    update_data: dict[str, Any] = {}
     for key, value in kwargs.items():
-        if key not in allowed_fields:
+        if key not in _ALLOWED_UPDATE_FIELDS:
             continue
-        if value is None and key not in nullable_fields:
+        if value is None and key not in _NULLABLE_UPDATE_FIELDS:
             continue
         update_data[key] = value
 
@@ -200,26 +213,46 @@ async def update_news_article(
     update_data['updated_at'] = datetime.now(UTC)
 
     await db.execute(update(NewsArticle).where(NewsArticle.id == article.id).values(**update_data))
-    await db.commit()
+    try:
+        await db.commit()
+    except IntegrityError:
+        await db.rollback()
+        raise
     await db.refresh(article)
 
     logger.info(
-        'Updated news article', article_id=article.id, slug=article.slug, updated_fields=list(update_data.keys())
+        'Updated news article',
+        article_id=article.id,
+        slug=article.slug,
+        updated_fields=list(update_data.keys()),
     )
     return article
 
 
-async def delete_news_article(db: AsyncSession, article: NewsArticle) -> bool:
+async def delete_news_article(db: AsyncSession, article: NewsArticle) -> None:
     """Delete a news article."""
-    await db.execute(delete(NewsArticle).where(NewsArticle.id == article.id))
+    # Capture fields before commit expires the ORM instance attributes
+    article_id = article.id
+    article_slug = article.slug
+
+    await db.execute(delete(NewsArticle).where(NewsArticle.id == article_id))
     await db.commit()
-    logger.info('Deleted news article', article_id=article.id, slug=article.slug)
-    return True
+
+    logger.info('Deleted news article', article_id=article_id, slug=article_slug)
 
 
-async def increment_views(db: AsyncSession, article_id: int) -> None:
-    """Increment the views counter for a news article (fire-and-forget)."""
-    await db.execute(
-        update(NewsArticle).where(NewsArticle.id == article_id).values(views_count=NewsArticle.views_count + 1)
+async def increment_views(db: AsyncSession, article_id: int) -> int:
+    """Atomically increment the views counter and return the new count.
+
+    Uses UPDATE … RETURNING so the caller can patch the ORM instance directly
+    without issuing a second SELECT (db.refresh).
+    """
+    result = await db.execute(
+        update(NewsArticle)
+        .where(NewsArticle.id == article_id)
+        .values(views_count=NewsArticle.views_count + 1)
+        .returning(NewsArticle.views_count)
     )
     await db.commit()
+    row = result.fetchone()
+    return row[0] if row else 0
diff --git a/app/database/models.py b/app/database/models.py
index 0580de02..33a1813d 100644
--- a/app/database/models.py
+++ b/app/database/models.py
@@ -3306,7 +3306,12 @@ class NewsArticle(Base):
 
     __tablename__ = 'news_articles'
     __table_args__ = (
+        # Covers the main public list query: WHERE is_published = true ORDER BY published_at DESC
         Index('ix_news_articles_published_at_published', 'is_published', 'published_at'),
+        # Covers the category-filtered public list: WHERE is_published = true AND category = ?
+        Index('ix_news_articles_published_category', 'is_published', 'category'),
+        # Covers the admin list query: ORDER BY created_at DESC
+        Index('ix_news_articles_created_at', 'created_at'),
     )
 
     id = Column(Integer, primary_key=True, index=True)
diff --git a/migrations/alembic/versions/0046_add_news_articles.py b/migrations/alembic/versions/0046_add_news_articles.py
index 0e9526a8..e5bb413c 100644
--- a/migrations/alembic/versions/0046_add_news_articles.py
+++ b/migrations/alembic/versions/0046_add_news_articles.py
@@ -7,15 +7,15 @@ Create Date: 2026-03-23
 Adds news_articles table for the cabinet news/blog feature.
 """
 
-from typing import Sequence, Union
+from collections.abc import Sequence
 
 import sqlalchemy as sa
 from alembic import op
 
 revision: str = '0046'
 down_revision: str | None = '0045'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
 
 
 def upgrade() -> None:
@@ -39,11 +39,31 @@ def upgrade() -> None:
         sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
         sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
     )
+
+    # Unique index on slug (model: unique=True, index=True)
     op.create_index('ix_news_articles_slug', 'news_articles', ['slug'], unique=True)
-    op.create_index('ix_news_articles_published_at', 'news_articles', ['published_at'])
+
+    # Composite: covers WHERE is_published = true ORDER BY published_at DESC
+    op.create_index(
+        'ix_news_articles_published_at_published',
+        'news_articles',
+        ['is_published', 'published_at'],
+    )
+
+    # Composite: covers WHERE is_published = true AND category = ?
+    op.create_index(
+        'ix_news_articles_published_category',
+        'news_articles',
+        ['is_published', 'category'],
+    )
+
+    # Covers admin list: ORDER BY created_at DESC
+    op.create_index('ix_news_articles_created_at', 'news_articles', ['created_at'])
 
 
 def downgrade() -> None:
-    op.drop_index('ix_news_articles_published_at', table_name='news_articles')
+    op.drop_index('ix_news_articles_created_at', table_name='news_articles')
+    op.drop_index('ix_news_articles_published_category', table_name='news_articles')
+    op.drop_index('ix_news_articles_published_at_published', table_name='news_articles')
     op.drop_index('ix_news_articles_slug', table_name='news_articles')
     op.drop_table('news_articles')