Phase 1: architectural fixes (concurrency, async/sync, data integrity)

Concurrency / performance
- Drop the global threading.RLock in DatabaseService. It was wrapping every
  _safe_execute and turning the 8-wide thread pool into a 1-wide one;
  Flask-SQLAlchemy's scoped_session already gives each thread its own
  session, so the lock was protecting nothing while serialising every DB
  op in the process.
- moderate_content now awaits _process_rules and _handle_no_matches; both
  push the blocking OpenAI call to asyncio.to_thread so the event loop
  stays responsive while waiting on the model.
- DiscordNotifier shares a single module-level requests.Session (instead
  of one per instance) and exposes an async wrapper that runs the POST
  via asyncio.to_thread, so the moderation request doesn't pin its loop
  on Discord latency / 429 retry backoff.
- load_user simplified to a single User.query.get(user_id). Previously it
  spun a fresh ThreadPoolExecutor and asyncio event loop on every
  authenticated request just to await an async wrapper around the same
  primary-key lookup.
- ResultCache now LRU-evicts the oldest entries when at capacity with no
  expired entries to free, instead of silently dropping new writes.
  _cleanup_expired_entries now sets _last_cleanup_time so the periodic
  throttle applies even when called from the hot-path fallback.

Data integrity
- APIUser gained UniqueConstraint(project_id, external_user_id).
  get_or_create_api_user catches IntegrityError, rolls back, and reads the
  winner's row so concurrent submits for the same external user no longer
  produce duplicate APIUser rows that split stats.
- ProjectMember gained UniqueConstraint(project_id, user_id). The
  invitation accept flow catches the IntegrityError and treats it as a
  benign duplicate-click.
- New scripts/dedup_before_unique_constraints.py to merge any pre-existing
  duplicate rows so the new constraints can be added cleanly. Dry-run by
  default, --apply to commit.
- Fix the API-user stats detached-instance bug: increment_api_user_stats
  in the DB service now does query + mutation + commit in one
  _safe_execute call. Previously the orchestrator mutated a detached ORM
  instance outside any session and the changes silently never persisted,
  so per-user counters drifted.
- Manual-review decision flow now commits the moderation result first,
  then atomically commits the APIUser stat bump in the same request
  context (instead of relying on the detached-instance pattern).

Correctness
- ContentType enum is now {text, markdown, html} to match the route's
  whitelist. Previously the schema declared text/image/video/audio while
  the route accepted text/markdown/html, so markdown/html submissions
  died with a Pydantic 400 before reaching the route.
- Project invitation emails are normalised to lowercase on store and
  compared case-insensitively on accept/decline. Mixed-case stored
  invitations were silently locking recipients out of accepting their
  own invite.
- Admin create_user lowercases the email and requires 8-char passwords
  (matching the public registration policy). The previous 6-char floor
  was weaker than the public path despite admin-created accounts often
  being privileged.
- AI moderator response parsing now uses (content or "").strip() — OpenAI
  returns content=None on safety refusals and the raw .strip() raised
  AttributeError that the surrounding except didn't catch.

Author: Bentlybro

app/__init__.py

@@ -200,34 +200,15 @@ def filtered_werkzeug_log(msg, *args, **kwargs):
         }
     )
 
-    # User loader for Flask-Login
+    # User loader for Flask-Login. Called on every authenticated request, so
+    # keep it cheap: a simple synchronous primary-key lookup inside the
+    # existing app context. The previous implementation spun up a fresh
+    # ThreadPoolExecutor AND a fresh asyncio event loop per call just to
+    # await an async method that ultimately wraps ``User.query.get(user_id)``.
     @login_manager.user_loader
     def load_user(user_id):
-        import asyncio
-        import concurrent.futures
-
-        from flask import current_app, has_app_context
-
-        from app.services.database_service import db_service
-
-        # Get the current app reference before creating the thread
-        if has_app_context():
-            app = current_app._get_current_object()
-        else:
-            # If no app context, return None (user not authenticated)
-            return None
-
-        def run_async_with_context():
-            def async_operation():
-                with app.app_context():
-                    return asyncio.run(db_service.get_user_by_id(user_id))
-
-            return async_operation()
-
-        # Run in a separate thread with proper app context
-        with concurrent.futures.ThreadPoolExecutor() as executor:
-            future = executor.submit(run_async_with_context)
-            return future.result()
+        from app.models.user import User
+        return User.query.get(user_id)
 
     # Initialize OAuth
     from app.routes.auth import oauth

app/models/api_user.py

@@ -8,6 +8,14 @@
 
 class APIUser(db.Model):
     __tablename__ = 'api_users'
+    # An external user is uniquely identified by (project_id, external_user_id).
+    # Without this constraint two concurrent /api/moderate requests for the
+    # same external user could both miss the SELECT and both INSERT, producing
+    # duplicate rows that split stats across them.
+    __table_args__ = (
+        db.UniqueConstraint('project_id', 'external_user_id',
+                            name='uq_api_users_project_external'),
+    )
 
     id = db.Column(db.String(36), primary_key=True,
                    default=lambda: str(uuid.uuid4()))

app/models/project.py

@@ -6,6 +6,13 @@
 
 class ProjectMember(db.Model):
     __tablename__ = 'project_members'
+    # A user can only hold one membership row per project. Without this
+    # constraint the accept_invitation flow is a TOCTOU race: two concurrent
+    # requests both pass is_member(user_id)=False and both INSERT.
+    __table_args__ = (
+        db.UniqueConstraint('project_id', 'user_id',
+                            name='uq_project_members_project_user'),
+    )
 
     id = db.Column(db.String(36), primary_key=True,
                    default=lambda: str(uuid.uuid4()))

app/routes/admin.py

@@ -164,7 +164,10 @@ async def create_user():
     """Create a new user"""
     try:
         username = request.form.get('username', '').strip()
-        email = request.form.get('email', '').strip()
+        # Lowercase to match the public registration / login flows so admins
+        # can't accidentally create case-duplicated accounts that
+        # get_user_by_email(email.lower()) then fetches non-deterministically.
+        email = request.form.get('email', '').strip().lower()
         password = request.form.get('password', '')
         confirm_password = request.form.get('confirm_password', '')
         is_admin = request.form.get('is_admin') == '1'
@@ -179,8 +182,11 @@ async def create_user():
             flash('Passwords do not match.', 'error')
             return redirect(url_for('admin.users'))
 
-        if len(password) < 6:
-            flash('Password must be at least 6 characters long.', 'error')
+        # Match the public registration policy (auth.py:_is_valid_password):
+        # 8 chars min. The previous 6-char floor was weaker than the public
+        # path despite admin-created accounts often being privileged.
+        if len(password) < 8:
+            flash('Password must be at least 8 characters long.', 'error')
             return redirect(url_for('admin.users'))
 
         # Check if username or email already exists

app/routes/dashboard.py

@@ -768,7 +768,10 @@ async def invite_member(project_id):
         flash('You do not have permission to invite members', 'error')
         return redirect(url_for('dashboard.project_members', project_id=project_id))
 
-    email = request.form.get('email')
+    # Normalise to lowercase so invitations match regardless of how the
+    # invitee typed their email at signup. Mixed-case stored invitations
+    # would silently lock the recipient out of accepting their own invite.
+    email = (request.form.get('email') or '').strip().lower()
     role = request.form.get('role', 'member')
 
     if not email:
@@ -916,8 +919,9 @@ async def accept_invitation(token):
         flash('Please log in to accept the invitation', 'info')
         return redirect(url_for('auth.login'))
 
-    # Check if user email matches invitation
-    if current_user.email != invitation.email:
+    # Case-insensitive comparison: invitations stored lowercase by invite_member,
+    # but defend against legacy mixed-case data on either side.
+    if (current_user.email or '').lower() != (invitation.email or '').lower():
         flash('This invitation was sent to a different email address', 'error')
         return redirect(url_for('dashboard.index'))
 
@@ -929,7 +933,10 @@ async def accept_invitation(token):
         flash('You are already a member of this project', 'info')
         return redirect(url_for('dashboard.project_detail', project_id=project.id))
 
-    # Add user as member
+    # Add user as member. The (project_id, user_id) unique constraint on
+    # ProjectMember races-safes this: if two accept clicks arrive at once and
+    # both passed the is_member() check above, only one INSERT commits; the
+    # other rolls back and we treat it as a benign duplicate.
     membership = ProjectMember(
         project_id=project.id,
         user_id=current_user.id,
@@ -938,7 +945,12 @@ async def accept_invitation(token):
 
     invitation.status = 'accepted'
     db.session.add(membership)
-    db.session.commit()
+    try:
+        db.session.commit()
+    except SQLAlchemyError:
+        db.session.rollback()
+        flash('You are already a member of this project', 'info')
+        return redirect(url_for('dashboard.project_detail', project_id=project.id))
 
     flash(
         f'You have successfully joined the project "{project.name}"', 'success')
@@ -955,7 +967,7 @@ async def decline_invitation(token):
         flash('This invitation is no longer valid', 'error')
         return redirect(url_for('dashboard.index'))
 
-    if current_user.email != invitation.email:
+    if (current_user.email or '').lower() != (invitation.email or '').lower():
         flash('This invitation was sent to a different email address', 'error')
         return redirect(url_for('dashboard.index'))
 

app/routes/manual_review.py

@@ -149,13 +149,16 @@ async def make_decision(content_id):
         )
         db.session.add(manual_result)
 
-        # Update API user stats if available
+        # Commit the content + moderation result first, then atomically bump
+        # API user stats in a separate transaction. Both happen in the same
+        # request context so we do them synchronously here.
+        db.session.commit()
+
         if content.api_user_id:
             api_user = APIUser.query.get(content.api_user_id)
             if api_user:
                 api_user.update_stats(decision)
-
-        db.session.commit()
+                db.session.commit()
 
         current_app.logger.info(
             f"Manual decision made on content {content_id}: {decision} by {current_user.username}")
@@ -238,7 +241,9 @@ async def bulk_decision():
                 )
                 db.session.add(manual_result)
 
-                # Update API user stats if available
+                # Update API user stats — mutates the attached instance
+                # inside this request's session; flushed by the single
+                # db.session.commit() below.
                 if content.api_user_id:
                     api_user = APIUser.query.get(content.api_user_id)
                     if api_user:

app/schemas/api_schemas.py

@@ -8,11 +8,16 @@
 
 
 class ContentType(str, Enum):
-    """Allowed content types for moderation"""
+    """Allowed content types for moderation.
+
+    Must match the whitelist in app/routes/api.py:moderate_content. Previously
+    the enum listed text/image/video/audio while the route only accepted
+    text/markdown/html, so markdown/html submissions hit Pydantic validation
+    and died with a 400 before reaching the route.
+    """
     TEXT = "text"
-    IMAGE = "image"
-    VIDEO = "video"
-    AUDIO = "audio"
+    MARKDOWN = "markdown"
+    HTML = "html"
 
 
 class ModerationStatus(str, Enum):

app/services/ai/ai_moderator.py

@@ -758,7 +758,12 @@ def make_api_call():
 
             response = self._retry_api_call(make_api_call)
 
-            result_text = response.choices[0].message.content.strip()
+            # OpenAI returns content=None when the model refuses to generate
+            # (safety filter, function-call response, etc). `.strip()` on None
+            # raises AttributeError which the surrounding except clauses do
+            # not catch — guard with `or ""` so the empty-string path is
+            # handled by the JSON parser fallback.
+            result_text = (response.choices[0].message.content or "").strip()
 
             # Parse JSON response
             try:
@@ -999,7 +1004,12 @@ def make_api_call():
 
             response = self._retry_api_call(make_api_call)
 
-            result_text = response.choices[0].message.content.strip()
+            # OpenAI returns content=None when the model refuses to generate
+            # (safety filter, function-call response, etc). `.strip()` on None
+            # raises AttributeError which the surrounding except clauses do
+            # not catch — guard with `or ""` so the empty-string path is
+            # handled by the JSON parser fallback.
+            result_text = (response.choices[0].message.content or "").strip()
 
             # Parse JSON response
             try:

app/services/ai/result_cache.py

@@ -73,23 +73,47 @@ def cache_result(self, cache_key, result):
             if len(ResultCache._shared_cache) >= ResultCache._max_cache_size:
                 self._aggressive_cleanup()
 
-            # Only cache if we have space or successfully cleaned up
-            if len(ResultCache._shared_cache) < ResultCache._max_cache_size:
-                ResultCache._shared_cache[cache_key] = {
-                    'result': result,
-                    'timestamp': time.time()
-                }
-
-                # Track stores for this request
-                ResultCache._current_request_stores += 1
-            else:
-                # Cache is full, log warning
-                current_app.logger.warning("Cache full, dropping new entry to prevent memory leak")
+            # If aggressive_cleanup couldn't free space (no expired entries),
+            # fall back to evicting the oldest entries by timestamp. The
+            # previous behaviour silently dropped new writes once the cache
+            # filled, which broke caching entirely under sustained load.
+            if len(ResultCache._shared_cache) >= ResultCache._max_cache_size:
+                self._evict_oldest(target_count=max(1, ResultCache._max_cache_size // 100))
+
+            ResultCache._shared_cache[cache_key] = {
+                'result': result,
+                'timestamp': time.time()
+            }
+
+            # Track stores for this request
+            ResultCache._current_request_stores += 1
 
             # Perform cleanup if we've reached the threshold
             if len(ResultCache._shared_cache) >= ResultCache._cleanup_threshold:
                 self._cleanup_expired_entries()
 
+    def _evict_oldest(self, target_count: int) -> int:
+        """Evict the ``target_count`` oldest entries by timestamp.
+
+        Used as a last-resort eviction when the cache is at capacity but no
+        entries have actually expired yet. Keeps the most recently cached
+        results (typically the most useful) and discards stale ones.
+        """
+        if not ResultCache._shared_cache:
+            return 0
+        sorted_keys = sorted(
+            ResultCache._shared_cache.items(),
+            key=lambda item: item[1]['timestamp'],
+        )
+        removed = 0
+        for key, _ in sorted_keys[:target_count]:
+            if ResultCache._shared_cache.pop(key, None) is not None:
+                removed += 1
+        if removed:
+            current_app.logger.info(
+                f"Cache LRU eviction: removed {removed} oldest entries to make room")
+        return removed
+
     def get_request_cache_summary(self):
         """Get summary of cache operations for current request"""
         stores = ResultCache._current_request_stores
@@ -140,6 +164,12 @@ def _cleanup_expired_entries(self):
         for key in expired_keys:
             ResultCache._shared_cache.pop(key, None)
 
+        # Always update _last_cleanup_time so the periodic-cleanup throttle
+        # applies whether we were called from the periodic check or from the
+        # hot-path cache_result fallback. Without this, hot-path writes after
+        # the threshold trigger an O(n) scan on every single insert.
+        ResultCache._last_cleanup_time = current_time
+
         if expired_keys:
             current_app.logger.info(f"Cleaned up {len(expired_keys)} expired cache entries")