WIP checkpoint: Text filtering service + profanity config

backend/alembic/versions/2a9b9873244a_add_presence_columns.py

@@ -1,29 +0,0 @@
-"""add presence columns to users
-
-Revision ID: 2a9b9873244a
-Revises: 0001_initial
-Create Date: 2026-08-18 15:59:09.298919
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '2a9b9873244a'
-down_revision: Union[str, None] = '0001_initial'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    """Add last_seen_at and status_message columns to users table."""
-    op.add_column('users', sa.Column('last_seen_at', sa.DateTime(timezone=True), nullable=True))
-    op.add_column('users', sa.Column('status_message', sa.Text(), nullable=True))
-
-
-def downgrade() -> None:
-    """Remove presence columns from users table."""
-    op.drop_column('users', 'status_message')
-    op.drop_column('users', 'last_seen_at')

backend/app/main.py

@@ -6,7 +6,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.config import settings
-from app.routers import auth, users, worlds, assets, presence
+from app.routers import auth, users, worlds, assets
 
 
 @asynccontextmanager
@@ -38,7 +38,6 @@ app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
 app.include_router(users.router, prefix="/api/users", tags=["users"])
 app.include_router(worlds.router, prefix="/api/worlds", tags=["worlds"])
 app.include_router(assets.router, prefix="/api/assets", tags=["assets"])
-app.include_router(presence.router, prefix="/api/presence", tags=["presence"])
 
 
 @app.get("/api/health")

backend/app/models/user.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text
+from sqlalchemy import Column, Integer, String, Boolean, DateTime
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
 
@@ -16,10 +16,6 @@ class User(Base):
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
 
-    # Presence / status
-    last_seen_at = Column(DateTime(timezone=True), nullable=True)
-    status_message = Column(Text, nullable=True)
-
     sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan")
     worlds = relationship("World", back_populates="owner")
     assets = relationship("Asset", back_populates="owner")

backend/app/routers/presence.py

@@ -1,155 +0,0 @@
-"""Presence & status router.
-
-Endpoints:
-- POST /api/presence/heartbeat  – record a heartbeat (sets Redis TTL key)
-- GET  /api/presence/presence   – query presence for a list of user ids
-- PATCH /api/presence/status    – update the authenticated user's status message
-"""
-
-from datetime import datetime, timezone
-from typing import Annotated
-
-from fastapi import APIRouter, Depends, HTTPException, status
-from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.config import settings
-from app.database import get_db
-from app.dependencies import CurrentUser
-from app.models.user import User
-from app.services.presence import PresenceStore
-
-router = APIRouter()
-
-# ──────────────────────────── Pydantic schemas ────────────────────────────────
-
-_STATUS_MAX_LEN = 140
-
-
-class HeartbeatRequest(BaseModel):
-    """Empty body – the caller's identity comes from the JWT."""
-
-
-class PresenceUser(BaseModel):
-    user_id: int
-    is_online: bool
-    last_seen_at: datetime | None
-    status_message: str | None
-
-
-class PresenceResponse(BaseModel):
-    users: list[PresenceUser]
-
-
-class StatusUpdate(BaseModel):
-    status_message: str | None = Field(default=None, max_length=_STATUS_MAX_LEN)
-
-    @field_validator("status_message")
-    @classmethod
-    def strip_message(cls, v: str | None) -> str | None:
-        if v is not None:
-            v = v.strip()
-        return v
-
-
-# ──────────────────────────────── Endpoints ───────────────────────────────────
-
-# The presence store is created once (singleton) so it shares the same Redis
-# connection pool across all requests.
-_presence_store = PresenceStore()
-
-
-@router.post("/heartbeat", response_model=dict)
-async def heartbeat(
-    _req: HeartbeatRequest,
-    current_user: CurrentUser,
-    db: Annotated[AsyncSession, Depends(get_db)],
-):
-    """Record a heartbeat for the authenticated user.
-
-    Sets a Redis key with a TTL so the server can derive online/offline
-    status without trusting client claims.  Also persists last_seen_at to
-    the database.
-    """
-    # Update Redis TTL (presence heartbeat)
-    await _presence_store.heartbeat(current_user.id)
-
-    # Persist last_seen_at to the database
-    now = datetime.now(timezone.utc)
-    result = await db.execute(
-        select(User).where(User.id == current_user.id)
-    )
-    user = result.scalar_one_or_none()
-    if user:
-        user.last_seen_at = now
-        await db.commit()
-
-    return {"status": "ok", "user_id": current_user.id}
-
-
-@router.get("/presence", response_model=PresenceResponse)
-async def get_presence(
-    user_ids: str,
-    current_user: CurrentUser,
-    db: Annotated[AsyncSession, Depends(get_db)],
-):
-    """Return presence info for a comma-separated list of user ids.
-
-    Online/offline is derived from the Redis heartbeat TTL.
-    last_seen_at and status_message come from the database.
-    """
-    ids = [int(uid.strip()) for uid in user_ids.split(",") if uid.strip()]
-    if not ids:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="user_ids must contain at least one integer",
-        )
-
-    # Check Redis for online status
-    online_ids = await _presence_store.is_online_batch(ids)
-
-    # Fetch DB records for last_seen_at and status_message
-    result = await db.execute(
-        select(User).where(User.id.in_(ids))
-    )
-    users = {u.id: u for u in result.scalars().all()}
-
-    presence_users = []
-    for uid in ids:
-        user = users.get(uid)
-        if user is None:
-            continue
-        presence_users.append(
-            PresenceUser(
-                user_id=uid,
-                is_online=uid in online_ids,
-                last_seen_at=user.last_seen_at,
-                status_message=user.status_message,
-            )
-        )
-
-    return PresenceResponse(users=presence_users)
-
-
-@router.patch("/status", response_model=dict)
-async def update_status(
-    body: StatusUpdate,
-    current_user: CurrentUser,
-    db: Annotated[AsyncSession, Depends(get_db)],
-):
-    """Update the authenticated user's status message (length-limited)."""
-    result = await db.execute(
-        select(User).where(User.id == current_user.id)
-    )
-    user = result.scalar_one_or_none()
-    if not user:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail="User not found",
-        )
-
-    user.status_message = body.status_message
-    await db.commit()
-
-    return {"status": "ok", "status_message": body.status_message}

backend/app/services/__init__.py

@@ -0,0 +1 @@
+"""Application services."""

backend/app/services/presence.py

@@ -1,196 +0,0 @@
-"""Redis-backed presence store.
-
-Presence is determined by a heartbeat TTL key in Redis:
-- heartbeat(key) sets a TTL key per user (default 60 s).
-- is_online(user_id) checks whether the key exists.
-- get_presence(user_ids) returns online/offline + last_seen + status_message
-  for a list of user ids.
-
-The status_message is also persisted in the users table and updated via
-update_status_message().
-"""
-
-from __future__ import annotations
-
-import logging
-from datetime import datetime, timezone
-from typing import Any
-
-import redis.asyncio as aioredis
-
-from app.config import settings
-
-logger = logging.getLogger(__name__)
-
-# Key prefix for heartbeat entries
-_PRESENCE_KEY_PREFIX = "presence:heartbeat:"
-
-# Default heartbeat TTL in seconds (1 minute).
-# Clients should heartbeat at least once per interval.
-_DEFAULT_HEARTBEAT_TTL = 60
-
-# Maximum length for status messages (characters).
-_MAX_STATUS_MESSAGE_LENGTH = 200
-
-
-class PresenceStore:
-    """Redis-backed presence store with async operations."""
-
-    def __init__(self, redis_url: str = settings.REDIS_URL, heartbeat_ttl: int = _DEFAULT_HEARTBEAT_TTL):
-        self._redis_url = redis_url
-        self._heartbeat_ttl = heartbeat_ttl
-        self._client: aioredis.Redis | None = None
-
-    # ------------------------------------------------------------------
-    # Lifecycle
-    # ------------------------------------------------------------------
-
-    async def connect(self) -> None:
-        """Connect to Redis."""
-        self._client = aioredis.from_url(
-            self._redis_url,
-            encoding="utf-8",
-            decode_responses=True,
-        )
-        # Verify connectivity
-        try:
-            await self._client.ping()
-        except Exception as exc:
-            logger.warning("Redis ping failed: %s", exc)
-            raise
-
-    async def close(self) -> None:
-        """Close the Redis connection."""
-        if self._client is not None:
-            await self._client.close()
-            self._client = None
-
-    @property
-    def client(self) -> aioredis.Redis:
-        """Return the Redis client, raising if not connected."""
-        if self._client is None:
-            raise RuntimeError("PresenceStore is not connected. Call connect() first.")
-        return self._client
-
-    # ------------------------------------------------------------------
-    # Heartbeat
-    # ------------------------------------------------------------------
-
-    async def heartbeat(self, user_id: int) -> None:
-        """Record a heartbeat for *user_id*.
-
-        Sets a TTL key so that the user is considered online as long as
-        heartbeats arrive at least once per heartbeat_ttl seconds.
-        """
-        key = self._heartbeat_key(user_id)
-        await self.client.set(key, "1", ex=self._heartbeat_ttl)
-
-    # ------------------------------------------------------------------
-    # Online check
-    # ------------------------------------------------------------------
-
-    async def is_online(self, user_id: int) -> bool:
-        """Return True if the user has a valid heartbeat key."""
-        key = self._heartbeat_key(user_id)
-        exists = await self.client.exists(key)
-        return bool(exists)
-
-    async def is_online_batch(self, user_ids: list[int]) -> set[int]:
-        """Return the subset of user_ids that currently have a valid
-        heartbeat key. Uses MGET (one round trip) rather than one EXISTS
-        per user."""
-        if not user_ids:
-            return set()
-        keys = [self._heartbeat_key(uid) for uid in user_ids]
-        values = await self.client.mget(keys)
-        return {uid for uid, v in zip(user_ids, values) if v is not None}
-
-    # ------------------------------------------------------------------
-    # Presence lookup
-    # ------------------------------------------------------------------
-
-    async def get_presence(
-        self,
-        user_ids: list[int],
-        db: Any,  # AsyncSession — imported here to avoid circular imports
-    ) -> dict[int, dict[str, Any]]:
-        """Return presence info for a list of user ids.
-
-        Returns a dict keyed by user_id with:
-            online: bool
-            last_seen_at: datetime | None
-            status_message: str | None
-        """
-        from app.models.user import User
-        from sqlalchemy import select
-
-        result = await db.execute(select(User).where(User.id.in_(user_ids)))
-        users = result.scalars().all()
-        user_map = {u.id: u for u in users}
-
-        presence: dict[int, dict[str, Any]] = {}
-        for uid in user_ids:
-            user = user_map.get(uid)
-            online = await self.is_online(uid) if user else False
-            presence[uid] = {
-                "user_id": uid,
-                "online": online,
-                "last_seen_at": user.last_seen_at if user else None,
-                "status_message": user.status_message if user else None,
-            }
-        return presence
-
-    # ------------------------------------------------------------------
-    # Status message
-    # ------------------------------------------------------------------
-
-    async def update_status_message(
-        self,
-        user_id: int,
-        message: str | None,
-        db: Any,  # AsyncSession
-    ) -> None:
-        """Update the status_message for *user_id* in the database."""
-        from app.models.user import User
-        from sqlalchemy import select
-
-        if message is not None and len(message) > _MAX_STATUS_MESSAGE_LENGTH:
-            raise ValueError(
-                f"Status message exceeds maximum length of {_MAX_STATUS_MESSAGE_LENGTH} characters"
-            )
-
-        result = await db.execute(select(User).where(User.id == user_id))
-        user = result.scalar_one_or_none()
-        if user is None:
-            raise ValueError(f"User {user_id} not found")
-
-        user.status_message = message
-        user.last_seen_at = datetime.now(timezone.utc)
-        await db.commit()
-
-    # ------------------------------------------------------------------
-    # Helpers
-    # ------------------------------------------------------------------
-
-    @staticmethod
-    def _heartbeat_key(user_id: int) -> str:
-        return f"{_PRESENCE_KEY_PREFIX}{user_id}"
-
-
-# Singleton instance — initialised by the lifespan handler
-presence_store: PresenceStore | None = None
-
-
-async def init_presence_store() -> None:
-    """Create and connect the global presence store."""
-    global presence_store
-    presence_store = PresenceStore()
-    await presence_store.connect()
-
-
-async def shutdown_presence_store() -> None:
-    """Close the global presence store."""
-    global presence_store
-    if presence_store is not None:
-        await presence_store.close()
-        presence_store = None

backend/app/services/text_filter.py

@@ -0,0 +1,340 @@
+"""Pluggable server-side text filtering service.
+
+Provides a ``TextFilter`` protocol with a deterministic default
+implementation (profanity/term list with normalization) and a hook
+for third-party providers.
+
+Usage::
+
+    from app.services.text_filter import TextFilter, FilterResult, \
+        ProfanityTextFilter, get_text_filter
+
+    # Default factory reads config from the path below
+    filt = get_text_filter()
+    result = filt.check("Hello wor1d!")
+    # result.action == FilterAction.ALLOW
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import unicodedata
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# Enums / Data classes
+# ---------------------------------------------------------------------------
+
+
+class FilterAction(str, Enum):
+    """Action the filter wants the caller to take."""
+
+    ALLOW = "allow"
+    REDACT = "redact"
+    REJECT = "reject"
+
+
+@dataclass(frozen=True)
+class FilterResult:
+    """Immutable result of a single ``TextFilter.check()`` call.
+
+    Attributes:
+        action: What to do with the message.
+        filtered_text: The text after redaction (may equal *text* if
+            action is ``ALLOW``).
+        matched_terms: Sorted list of terms that triggered the filter.
+    """
+
+    action: FilterAction
+    filtered_text: str
+    matched_terms: list[str] = field(default_factory=list)
+
+    @property
+    def is_allowed(self) -> bool:
+        return self.action == FilterAction.ALLOW
+
+    @property
+    def is_rejected(self) -> bool:
+        return self.action == FilterAction.REJECT
+
+    @property
+    def is_redacted(self) -> bool:
+        return self.action == FilterAction.REDACT
+
+
+# ---------------------------------------------------------------------------
+# Protocol / abstract base
+# ---------------------------------------------------------------------------
+
+
+class TextFilter(ABC):
+    """Protocol for server-side text filters.
+
+    Every concrete filter must implement ``check()``.  The default
+    implementation (``ProfanityTextFilter``) is deterministic and
+    requires no external service.  A third-party provider can be
+    plugged in by subclassing ``TextFilter`` and overriding ``check``.
+    """
+
+    @abstractmethod
+    def check(self, text: str) -> FilterResult:
+        """Analyse *text* and return a ``FilterResult``.
+
+        Args:
+            text: Raw message text to filter.
+
+        Returns:
+            A ``FilterResult`` with the decision and any redacted text.
+        """
+
+
+# ---------------------------------------------------------------------------
+# Normalisation helpers
+# ---------------------------------------------------------------------------
+
+# Common leetspeak substitutions (char -> set of possible originals)
+_LEETSPEAK: dict[str, str] = {
+    "0": "o",
+    "1": "i",
+    "3": "e",
+    "4": "a",
+    "5": "s",
+    "7": "t",
+    "8": "b",
+    "9": "g",
+    "@": "a",
+}
+
+# Normalise separators that are often used to obfuscate words
+_SEPARATORS = re.compile(r"[\s_\-\.]+")
+
+
+def _normalize(text: str) -> str:
+    """Case-fold, strip diacritics, and replace leetspeak / separators.
+
+    This produces a canonical form for matching against the term list.
+    """
+    # Case-fold
+    text = text.casefold()
+    # Strip diacritics (e.g. é -> e)
+    text = unicodedata.normalize("NFKD", text)
+    text = "".join(c for c in text if not unicodedata.combining(c))
+    # Replace leetspeak
+    normalized = ""
+    for ch in text:
+        normalized += _LEETSPEAK.get(ch, ch)
+    # Collapse separators
+    normalized = _SEPARATORS.sub("", normalized)
+    return normalized
+
+
+# ---------------------------------------------------------------------------
+# Default implementation
+# ---------------------------------------------------------------------------
+
+# Default config path (relative to project root)
+_DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent.parent.parent / "data" / "text_filter_config.json"
+
+
+@dataclass
+class _FilterPolicy:
+    """Filtering policy loaded from config."""
+
+    mode: str  # "redact" or "reject"
+    redaction_char: str = "*"
+    # Minimum term length to consider (avoids matching single chars)
+    min_term_length: int = 3
+    # If True, match terms case-insensitively after normalisation
+    case_insensitive: bool = True
+    # If True, match whole words only (after normalisation)
+    whole_word: bool = True
+
+
+class ProfanityTextFilter(TextFilter):
+    """Deterministic text filter backed by a JSON term list.
+
+    The config file is a JSON object with:
+
+    .. code-block:: json
+
+        {
+            "mode": "redact",
+            "redaction_char": "*",
+            "min_term_length": 3,
+            "case_insensitive": true,
+            "whole_word": true,
+            "terms": ["badword", "another", ...]
+        }
+
+    Terms are matched after normalisation (case-fold, leetspeak
+    substitution, separator removal, diacritic stripping).
+    """
+
+    def __init__(
+        self,
+        config_path: str | Path | None = None,
+        terms: list[str] | None = None,
+        policy: _FilterPolicy | None = None,
+    ) -> None:
+        """
+        Args:
+            config_path: Path to a JSON config file.  Ignored if *terms*
+                is provided.
+            terms: Inline list of terms (overrides config file).
+            policy: Filter policy.  Defaults to a redact policy.
+        """
+        if terms is not None:
+            self._terms: list[str] = [t.lower() for t in terms]
+        elif config_path is not None:
+            self._terms = self._load_terms(Path(config_path))
+        else:
+            # Try default path
+            if _DEFAULT_CONFIG_PATH.exists():
+                self._terms = self._load_terms(_DEFAULT_CONFIG_PATH)
+            else:
+                self._terms = []
+
+        self._policy = policy or _FilterPolicy(mode="redact")
+        # Pre-compile patterns for each term
+        self._patterns: list[tuple[re.Pattern[str], str]] = []
+        for term in self._terms:
+            if len(term) < self._policy.min_term_length:
+                continue
+            if self._policy.whole_word:
+                pattern = re.compile(
+                    r"(?<!\w)" + re.escape(term) + r"(?!\w)",
+                    re.IGNORECASE if self._policy.case_insensitive else 0,
+                )
+            else:
+                pattern = re.compile(
+                    re.escape(term),
+                    re.IGNORECASE if self._policy.case_insensitive else 0,
+                )
+            self._patterns.append((pattern, term))
+
+    # -- config loading --------------------------------------------------
+
+    @staticmethod
+    def _load_terms(config_path: Path) -> list[str]:
+        """Load terms from a JSON config file."""
+        with open(config_path, "r", encoding="utf-8") as f:
+            data: dict[str, Any] = json.load(f)
+
+        terms: list[str] = []
+        for raw in data.get("terms", []):
+            term = str(raw).strip()
+            if term:
+                terms.append(term.casefold())
+        return terms
+
+    # -- public API ------------------------------------------------------
+
+    def check(self, text: str) -> FilterResult:
+        """Check *text* against the loaded term list.
+
+        Returns a ``FilterResult`` with the appropriate action,
+        filtered text, and matched terms.
+        """
+        if not text:
+            return FilterResult(
+                action=FilterAction.ALLOW,
+                filtered_text=text,
+                matched_terms=[],
+            )
+
+        normalized = _normalize(text)
+        matched_terms: list[str] = []
+
+        # Collect all matches
+        for pattern, term in self._patterns:
+            if pattern.search(normalized):
+                if term not in matched_terms:
+                    matched_terms.append(term)
+
+        if not matched_terms:
+            return FilterResult(
+                action=FilterAction.ALLOW,
+                filtered_text=text,
+                matched_terms=[],
+            )
+
+        matched_terms.sort()
+
+        # Build filtered text by replacing matched terms
+        filtered = text
+        for term in matched_terms:
+            norm_term = _normalize(term)
+            # Find all occurrences in the normalised text and replace in original
+            # We need to map positions from normalised text back to original
+            filtered = self._replace_terms(filtered, term, norm_term)
+
+        action = FilterAction.REJECT if self._policy.mode == "reject" else FilterAction.REDACT
+
+        return FilterResult(
+            action=action,
+            filtered_text=filtered,
+            matched_terms=matched_terms,
+        )
+
+    def _replace_terms(
+        self, text: str, term: str, norm_term: str
+    ) -> str:
+        """Replace occurrences of *term* in *text* with redaction chars.
+
+        Uses normalised text for matching but preserves original casing
+        in the output.
+        """
+        norm_text = _normalize(text)
+        norm_term_len = len(norm_term)
+        result_chars = list(text)
+
+        start = 0
+        while start < len(norm_text):
+            idx = norm_text.find(norm_term, start)
+            if idx == -1:
+                break
+            # Replace characters in original text
+            for i in range(idx, min(idx + norm_term_len, len(result_chars))):
+                result_chars[i] = self._policy.redaction_char
+            start = idx + norm_term_len
+
+        return "".join(result_chars)
+
+    @property
+    def terms(self) -> list[str]:
+        """Read-only access to the loaded term list."""
+        return list(self._terms)
+
+    @property
+    def policy(self) -> _FilterPolicy:
+        """Read-only access to the filter policy."""
+        return self._policy
+
+
+# ---------------------------------------------------------------------------
+# Factory
+# ---------------------------------------------------------------------------
+
+
+def get_text_filter(
+    config_path: str | Path | None = None,
+    terms: list[str] | None = None,
+    policy: _FilterPolicy | None = None,
+) -> TextFilter:
+    """Return a ``TextFilter`` instance.
+
+    By default this creates a ``ProfanityTextFilter``.  To use a
+    third-party provider, subclass ``TextFilter`` and pass an
+    instance here instead.
+    """
+    return ProfanityTextFilter(
+        config_path=config_path,
+        terms=terms,
+        policy=policy,
+    )