refactor(core): comprehensive code quality improvements (#4127)

## Summary

This PR implements comprehensive code quality improvements based on a
systematic audit. The changes address critical, high, and medium
priority issues across the codebase.

### Critical Fixes
- **Thread-safe database initialization**: Added `asyncio.Lock` and
`threading.Lock` for safe lazy initialization in
`core/database/connection.py`
- **Configurable Claude model**: Moved hardcoded model name to
`core/config.py` settings

### High Priority Improvements
- **Filter logic refactoring** (`api/routers/plots.py`): Extracted
duplicated 11-category if/elif chains (repeated 4x) into a unified
dispatch pattern, reducing ~200 lines to ~50 lines
- **Accessibility fixes** (`FilterBar.tsx`): Added WCAG 2.1 compliant
`role="button"`, `tabIndex`, `aria-label`, and keyboard event handlers
- **Exception logging** (`api/routers/proxy.py`): Added logging for URL
validation failures
- **Task error handling** (`api/analytics.py`): Added
`add_done_callback` to fire-and-forget analytics tasks
- **Atomic upserts** (`core/database/repositories.py`): Changed from
non-atomic get+create/update to PostgreSQL `INSERT ... ON CONFLICT`
- **Field validation** (`core/database/repositories.py`): Added
`SPEC_UPDATABLE_FIELDS`, `LIBRARY_UPDATABLE_FIELDS`,
`IMPL_UPDATABLE_FIELDS` to prevent modification of internal fields

### Medium Priority Improvements
- **Type hints** (`api/schemas.py`): Updated to Python 3.10+ syntax
(`str | None` instead of `Optional[str]`)
- **Database constraints** (`core/database/models.py`): Added
`CheckConstraint` for `quality_score` (0-100 range) and `review_verdict`
(APPROVED/REJECTED enum)
- **Sync script helpers** (`automation/scripts/sync_to_postgres.py`):
Extracted `_parse_markdown_section()` helper, added validation for
spec_id format, quality_score range, and review_verdict values

## Files Changed
| File | Changes |
|------|---------|
| `core/database/connection.py` | Thread-safe lazy initialization |
| `core/config.py` | Claude model configuration |
| `core/generators/plot_generator.py` | Use configurable model settings
|
| `api/routers/plots.py` | Filter dispatch pattern refactor |
| `app/src/components/FilterBar.tsx` | Accessibility improvements |
| `api/routers/proxy.py` | Exception logging |
| `api/analytics.py` | Task error handling |
| `core/database/repositories.py` | Atomic upserts + field validation |
| `api/schemas.py` | Type hint cleanup |
| `core/database/models.py` | Database constraints |
| `automation/scripts/sync_to_postgres.py` | Helper extraction +
validation |

## Test plan
- [x] Ruff linting passes
- [x] Ruff formatting applied
- [ ] Unit tests pass (CI will validate)
- [ ] Integration tests pass (CI will validate)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: MarkusNeusinger

api/analytics.py

@@ -124,6 +124,17 @@ async def _send_plausible_event(user_agent: str, client_ip: str, name: str, url:
         logger.debug(f"Plausible tracking failed (non-critical): {e}")
 
 
+def _handle_task_exception(task: asyncio.Task) -> None:
+    """Handle exceptions from fire-and-forget tasks to prevent silent failures."""
+    try:
+        # This will re-raise any exception that occurred in the task
+        task.result()
+    except asyncio.CancelledError:
+        pass  # Task was cancelled, not an error
+    except Exception as e:
+        logger.warning(f"Background analytics task failed: {e}")
+
+
 def track_og_image(
     request: Request,
     page: str,
@@ -170,5 +181,6 @@ def track_og_image(
         for key, value in filters.items():
             props[f"filter_{key}"] = value
 
-    # Fire-and-forget: create task without awaiting
-    asyncio.create_task(_send_plausible_event(user_agent, client_ip, "og_image_view", url, props))
+    # Fire-and-forget: create task without awaiting, but add exception handler
+    task = asyncio.create_task(_send_plausible_event(user_agent, client_ip, "og_image_view", url, props))
+    task.add_done_callback(_handle_task_exception)

api/routers/plots.py

@@ -1,6 +1,7 @@
 """Filter endpoint for plots."""
 
 import logging
+from collections.abc import Callable
 
 from fastapi import APIRouter, Depends, Request
 from sqlalchemy.exc import SQLAlchemyError
@@ -19,6 +20,80 @@
 router = APIRouter(tags=["plots"])
 
 
+# =============================================================================
+# Filter Category Extractors - Unified dispatch pattern for filter logic
+# =============================================================================
+# These extractors define how to get filter-matching values for each category.
+# Each extractor takes (library, spec_tags, impl_tags) and returns a list of values.
+
+FilterExtractor = Callable[[str, dict, dict], list[str]]
+
+# Spec-level category extractors (match against spec tags)
+_SPEC_EXTRACTORS: dict[str, Callable[[dict], list[str]]] = {
+    "plot": lambda tags: tags.get("plot_type", []),
+    "data": lambda tags: tags.get("data_type", []),
+    "dom": lambda tags: tags.get("domain", []),
+    "feat": lambda tags: tags.get("features", []),
+}
+
+# Impl-level category extractors (match against impl tags)
+_IMPL_EXTRACTORS: dict[str, Callable[[dict], list[str]]] = {
+    "dep": lambda tags: tags.get("dependencies", []),
+    "tech": lambda tags: tags.get("techniques", []),
+    "pat": lambda tags: tags.get("patterns", []),
+    "prep": lambda tags: tags.get("dataprep", []),
+    "style": lambda tags: tags.get("styling", []),
+}
+
+
+def _get_category_values(category: str, spec_id: str, library: str, spec_tags: dict, impl_tags: dict) -> list[str]:
+    """
+    Get the values for a category from the appropriate tag source.
+
+    This unified function replaces the repeated if/elif chains throughout the module.
+
+    Args:
+        category: Filter category (lib, spec, plot, data, dom, feat, dep, tech, pat, prep, style)
+        spec_id: Specification ID
+        library: Library ID
+        spec_tags: Spec-level tags dict
+        impl_tags: Implementation-level tags dict
+
+    Returns:
+        List of values that match this category for the given image/spec/impl
+    """
+    if category == "lib":
+        return [library]
+    if category == "spec":
+        return [spec_id]
+    if category in _SPEC_EXTRACTORS:
+        return _SPEC_EXTRACTORS[category](spec_tags)
+    if category in _IMPL_EXTRACTORS:
+        return _IMPL_EXTRACTORS[category](impl_tags)
+    return []
+
+
+def _category_matches_filter(
+    category: str, values: list[str], spec_id: str, library: str, spec_tags: dict, impl_tags: dict
+) -> bool:
+    """
+    Check if any of the filter values match the category's values.
+
+    Args:
+        category: Filter category
+        values: Filter values to match against
+        spec_id: Specification ID
+        library: Library ID
+        spec_tags: Spec-level tags dict
+        impl_tags: Implementation-level tags dict
+
+    Returns:
+        True if any filter value matches, False otherwise
+    """
+    category_values = _get_category_values(category, spec_id, library, spec_tags, impl_tags)
+    return any(v in category_values for v in values)
+
+
 def _image_matches_groups(spec_id: str, library: str, groups: list[dict], spec_lookup: dict, impl_lookup: dict) -> bool:
     """Check if an image matches a set of filter groups."""
     if spec_id not in spec_lookup:
@@ -30,55 +105,22 @@ def _image_matches_groups(spec_id: str, library: str, groups: list[dict], spec_l
         category = group["category"]
         values = group["values"]
 
-        if category == "lib":
-            if library not in values:
-                return False
-        elif category == "spec":
-            if spec_id not in values:
-                return False
-        elif category == "plot":
-            spec_plot_types = spec_tags.get("plot_type", [])
-            if not any(v in spec_plot_types for v in values):
-                return False
-        elif category == "data":
-            spec_data_types = spec_tags.get("data_type", [])
-            if not any(v in spec_data_types for v in values):
-                return False
-        elif category == "dom":
-            spec_domains = spec_tags.get("domain", [])
-            if not any(v in spec_domains for v in values):
-                return False
-        elif category == "feat":
-            spec_features = spec_tags.get("features", [])
-            if not any(v in spec_features for v in values):
-                return False
-        # Impl-level tag filters (issue #2434)
-        elif category == "dep":
-            impl_deps = impl_tags.get("dependencies", [])
-            if not any(v in impl_deps for v in values):
-                return False
-        elif category == "tech":
-            impl_techs = impl_tags.get("techniques", [])
-            if not any(v in impl_techs for v in values):
-                return False
-        elif category == "pat":
-            impl_pats = impl_tags.get("patterns", [])
-            if not any(v in impl_pats for v in values):
-                return False
-        elif category == "prep":
-            impl_preps = impl_tags.get("dataprep", [])
-            if not any(v in impl_preps for v in values):
-                return False
-        elif category == "style":
-            impl_styles = impl_tags.get("styling", [])
-            if not any(v in impl_styles for v in values):
-                return False
+        if not _category_matches_filter(category, values, spec_id, library, spec_tags, impl_tags):
+            return False
     return True
 
 
-def _calculate_global_counts(all_specs: list) -> dict:
-    """Calculate global counts for all filter categories."""
-    global_counts: dict = {
+def _increment_category_counts(counts: dict, spec_id: str, library: str, spec_tags: dict, impl_tags: dict) -> None:
+    """Increment counts for all categories based on an image's spec/impl tags."""
+    all_categories = ["lib", "spec", "plot", "data", "dom", "feat", "dep", "tech", "pat", "prep", "style"]
+    for category in all_categories:
+        for value in _get_category_values(category, spec_id, library, spec_tags, impl_tags):
+            counts[category][value] = counts[category].get(value, 0) + 1
+
+
+def _create_empty_counts() -> dict:
+    """Create an empty counts dictionary with all categories initialized."""
+    return {
         "lib": {},
         "spec": {},
         "plot": {},
@@ -93,6 +135,18 @@ def _calculate_global_counts(all_specs: list) -> dict:
         "style": {},
     }
 
+
+def _sort_counts(counts: dict) -> dict:
+    """Sort counts by value descending, then key ascending."""
+    for category in counts:
+        counts[category] = dict(sorted(counts[category].items(), key=lambda x: (-x[1], x[0])))
+    return counts
+
+
+def _calculate_global_counts(all_specs: list) -> dict:
+    """Calculate global counts for all filter categories."""
+    global_counts = _create_empty_counts()
+
     for spec_obj in all_specs:
         if not spec_obj.impls:
             continue
@@ -102,104 +156,25 @@ def _calculate_global_counts(all_specs: list) -> dict:
             if not impl.preview_url:
                 continue
 
-            # Count library
-            global_counts["lib"][impl.library_id] = global_counts["lib"].get(impl.library_id, 0) + 1
-
-            # Count spec ID
-            global_counts["spec"][spec_obj.id] = global_counts["spec"].get(spec_obj.id, 0) + 1
-
-            # Count spec-level tags
-            for plot_type in spec_tags.get("plot_type", []):
-                global_counts["plot"][plot_type] = global_counts["plot"].get(plot_type, 0) + 1
-
-            for data_type in spec_tags.get("data_type", []):
-                global_counts["data"][data_type] = global_counts["data"].get(data_type, 0) + 1
-
-            for domain in spec_tags.get("domain", []):
-                global_counts["dom"][domain] = global_counts["dom"].get(domain, 0) + 1
-
-            for feature in spec_tags.get("features", []):
-                global_counts["feat"][feature] = global_counts["feat"].get(feature, 0) + 1
-
-            # Count impl-level tags (issue #2434)
             impl_tags = impl.impl_tags or {}
-            for dep in impl_tags.get("dependencies", []):
-                global_counts["dep"][dep] = global_counts["dep"].get(dep, 0) + 1
-            for tech in impl_tags.get("techniques", []):
-                global_counts["tech"][tech] = global_counts["tech"].get(tech, 0) + 1
-            for pat in impl_tags.get("patterns", []):
-                global_counts["pat"][pat] = global_counts["pat"].get(pat, 0) + 1
-            for prep in impl_tags.get("dataprep", []):
-                global_counts["prep"][prep] = global_counts["prep"].get(prep, 0) + 1
-            for style in impl_tags.get("styling", []):
-                global_counts["style"][style] = global_counts["style"].get(style, 0) + 1
-
-    # Sort counts
-    for category in global_counts:
-        global_counts[category] = dict(sorted(global_counts[category].items(), key=lambda x: (-x[1], x[0])))
+            _increment_category_counts(global_counts, spec_obj.id, impl.library_id, spec_tags, impl_tags)
 
-    return global_counts
+    return _sort_counts(global_counts)
 
 
 def _calculate_contextual_counts(filtered_images: list[dict], spec_id_to_tags: dict, impl_lookup: dict) -> dict:
     """Calculate contextual counts from filtered images."""
-    counts: dict = {
-        "lib": {},
-        "spec": {},
-        "plot": {},
-        "data": {},
-        "dom": {},
-        "feat": {},
-        # Impl-level tag counts (issue #2434)
-        "dep": {},
-        "tech": {},
-        "pat": {},
-        "prep": {},
-        "style": {},
-    }
+    counts = _create_empty_counts()
 
     for img in filtered_images:
         spec_id = img["spec_id"]
         library = img["library"]
         spec_tags = spec_id_to_tags.get(spec_id, {})
         impl_tags = impl_lookup.get((spec_id, library), {})
 
-        # Count library
-        counts["lib"][library] = counts["lib"].get(library, 0) + 1
-
-        # Count spec ID
-        counts["spec"][spec_id] = counts["spec"].get(spec_id, 0) + 1
-
-        # Count spec-level tags
-        for plot_type in spec_tags.get("plot_type", []):
-            counts["plot"][plot_type] = counts["plot"].get(plot_type, 0) + 1
-
-        for data_type in spec_tags.get("data_type", []):
-            counts["data"][data_type] = counts["data"].get(data_type, 0) + 1
+        _increment_category_counts(counts, spec_id, library, spec_tags, impl_tags)
 
-        for domain in spec_tags.get("domain", []):
-            counts["dom"][domain] = counts["dom"].get(domain, 0) + 1
-
-        for feature in spec_tags.get("features", []):
-            counts["feat"][feature] = counts["feat"].get(feature, 0) + 1
-
-        # Count impl-level tags (issue #2434)
-        for dep in impl_tags.get("dependencies", []):
-            counts["dep"][dep] = counts["dep"].get(dep, 0) + 1
-        for tech in impl_tags.get("techniques", []):
-            counts["tech"][tech] = counts["tech"].get(tech, 0) + 1
-        for pat in impl_tags.get("patterns", []):
-            counts["pat"][pat] = counts["pat"].get(pat, 0) + 1
-        for prep in impl_tags.get("dataprep", []):
-            counts["prep"][prep] = counts["prep"].get(prep, 0) + 1
-        for style in impl_tags.get("styling", []):
-            counts["style"][style] = counts["style"].get(style, 0) + 1
-
-    # Sort counts
-    for category in counts:
-        counts[category] = dict(sorted(counts[category].items(), key=lambda x: (-x[1], x[0])))
-
-    return counts
+    return _sort_counts(counts)
 
 
 def _calculate_or_counts(
@@ -240,38 +215,9 @@ def _calculate_or_counts(
             spec_tags = spec_id_to_tags.get(spec_id, {})
             impl_tags = impl_lookup.get((spec_id, library), {})
 
-            if category == "lib":
-                group_counts[library] = group_counts.get(library, 0) + 1
-            elif category == "spec":
-                group_counts[spec_id] = group_counts.get(spec_id, 0) + 1
-            elif category == "plot":
-                for v in spec_tags.get("plot_type", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            elif category == "data":
-                for v in spec_tags.get("data_type", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            elif category == "dom":
-                for v in spec_tags.get("domain", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            elif category == "feat":
-                for v in spec_tags.get("features", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            # Impl-level tag counts (issue #2434)
-            elif category == "dep":
-                for v in impl_tags.get("dependencies", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            elif category == "tech":
-                for v in impl_tags.get("techniques", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            elif category == "pat":
-                for v in impl_tags.get("patterns", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            elif category == "prep":
-                for v in impl_tags.get("dataprep", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
-            elif category == "style":
-                for v in impl_tags.get("styling", []):
-                    group_counts[v] = group_counts.get(v, 0) + 1
+            # Use unified value extractor
+            for value in _get_category_values(category, spec_id, library, spec_tags, impl_tags):
+                group_counts[value] = group_counts.get(value, 0) + 1
 
         # Sort by count descending
         group_counts = dict(sorted(group_counts.items(), key=lambda x: (-x[1], x[0])))

api/routers/proxy.py

@@ -1,13 +1,16 @@
 """HTML proxy endpoint for interactive plots with size reporting."""
 
 import json
+import logging
 from urllib.parse import urlparse
 
 import httpx
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import HTMLResponse
 
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(tags=["proxy"])
 
 # Allowed origins for postMessage
@@ -135,7 +138,8 @@ def build_safe_gcs_url(url: str) -> str | None:
         # Reconstruct URL from hardcoded values to prevent SSRF
         # This breaks the taint flow by not using the original URL
         return f"https://{ALLOWED_HOST}/{safe_path}"
-    except Exception:
+    except Exception as e:
+        logger.warning("URL validation failed for %s: %s", url, e)
         return None