feat: move sensitive images to moderation service (#544) (#644)

consolidates sensitive image management in the rust moderation service:

- adds sensitive_images table to moderation db with migration
- adds GET /sensitive-images (public), POST /admin/sensitive-images,
POST /admin/sensitive-images/remove endpoints to moderation service
- adds get_sensitive_images() method to ModerationClient
- updates backend /moderation/sensitive-images to proxy to moderation service
- adds migration script for existing data

closes #544

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

Co-authored-by: Claude Opus 4.5 <[e-mail]>

backend/src/backend/_internal/moderation_client.py

···
     error: str | None = None
 
 
+@dataclass
+class SensitiveImagesResult:
+    """result from fetching sensitive images."""
+
+    image_ids: list[str]
+    urls: list[str]
+
+
 class ModerationClient:
     """client for the plyr.fm moderation service.
 
···
         except Exception as e:
             logger.warning("failed to emit copyright label for %s: %s", uri, e)
             return EmitLabelResult(success=False, error=str(e))
+
+    async def get_sensitive_images(self) -> SensitiveImagesResult:
+        """fetch all sensitive images from the moderation service.
+
+        returns:
+            SensitiveImagesResult with image_ids and urls
+
+        raises:
+            httpx.HTTPStatusError: on non-2xx response
+            httpx.TimeoutException: on timeout
+        """
+        async with httpx.AsyncClient(timeout=self.timeout) as client:
+            response = await client.get(
+                f"{self.labeler_url}/sensitive-images",
+                # no auth required for this public endpoint
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            return SensitiveImagesResult(
+                image_ids=data.get("image_ids", []),
+                urls=data.get("urls", []),
+            )
 
     async def get_active_labels(self, uris: list[str]) -> set[str]:
         """check which URIs have active (non-negated) copyright-violation labels.

backend/src/backend/api/moderation.py

···
 """content moderation api endpoints."""
 
-from typing import Annotated
+import logging
 
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Request
 from pydantic import BaseModel
-from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from backend.models import SensitiveImage, get_db
+from backend._internal.moderation_client import get_moderation_client
 from backend.utilities.rate_limit import limiter
+
+logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/moderation", tags=["moderation"])
 
···
 @limiter.limit("10/minute")
 async def get_sensitive_images(
     request: Request,
-    db: Annotated[AsyncSession, Depends(get_db)],
 ) -> SensitiveImagesResponse:
     """get all flagged sensitive images.
 
+    proxies to the moderation service which is the source of truth
+    for sensitive image data.
+
     returns both image_ids (for R2-stored images) and full URLs
     (for external images like avatars). clients should check both.
     """
-    result = await db.execute(select(SensitiveImage))
-    images = result.scalars().all()
+    client = get_moderation_client()
+    result = await client.get_sensitive_images()
 
-    image_ids = [img.image_id for img in images if img.image_id]
-    urls = [img.url for img in images if img.url]
-
-    return SensitiveImagesResponse(image_ids=image_ids, urls=urls)
+    return SensitiveImagesResponse(
+        image_ids=result.image_ids,
+        urls=result.urls,
+    )

backend/tests/test_moderation.py

···
 
 import httpx
 import pytest
+from fastapi.testclient import TestClient
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
···
     get_active_copyright_labels,
     scan_track_for_copyright,
 )
-from backend._internal.moderation_client import ModerationClient, ScanResult
+from backend._internal.moderation_client import (
+    ModerationClient,
+    ScanResult,
+    SensitiveImagesResult,
+)
 from backend.models import Artist, CopyrightScan, Track
 
 
···
 
     # scan2 should still be flagged
     assert scan2.is_flagged is True
+
+
+# tests for sensitive images
+
+
+async def test_moderation_client_get_sensitive_images() -> None:
+    """test ModerationClient.get_sensitive_images() with successful response."""
+    mock_response = Mock()
+    mock_response.json.return_value = {
+        "image_ids": ["abc123", "def456"],
+        "urls": ["https://example.com/image.jpg"],
+    }
+    mock_response.raise_for_status.return_value = None
+
+    client = ModerationClient(
+        service_url="https://test.example.com",
+        labeler_url="https://labeler.example.com",
+        auth_token="test-token",
+        timeout_seconds=30,
+        label_cache_prefix="test:label:",
+        label_cache_ttl_seconds=300,
+    )
+
+    with patch("httpx.AsyncClient.get", new_callable=AsyncMock) as mock_get:
+        mock_get.return_value = mock_response
+
+        result = await client.get_sensitive_images()
+
+    assert result.image_ids == ["abc123", "def456"]
+    assert result.urls == ["https://example.com/image.jpg"]
+    mock_get.assert_called_once()
+
+
+async def test_moderation_client_get_sensitive_images_empty() -> None:
+    """test ModerationClient.get_sensitive_images() with empty response."""
+    mock_response = Mock()
+    mock_response.json.return_value = {"image_ids": [], "urls": []}
+    mock_response.raise_for_status.return_value = None
+
+    client = ModerationClient(
+        service_url="https://test.example.com",
+        labeler_url="https://labeler.example.com",
+        auth_token="test-token",
+        timeout_seconds=30,
+        label_cache_prefix="test:label:",
+        label_cache_ttl_seconds=300,
+    )
+
+    with patch("httpx.AsyncClient.get", new_callable=AsyncMock) as mock_get:
+        mock_get.return_value = mock_response
+
+        result = await client.get_sensitive_images()
+
+    assert result.image_ids == []
+    assert result.urls == []
+
+
+async def test_get_sensitive_images_endpoint(
+    client: TestClient,
+) -> None:
+    """test GET /moderation/sensitive-images endpoint proxies to moderation service."""
+    mock_result = SensitiveImagesResult(
+        image_ids=["image1", "image2"],
+        urls=["https://example.com/avatar.jpg"],
+    )
+
+    with patch("backend.api.moderation.get_moderation_client") as mock_get_client:
+        mock_client = AsyncMock()
+        mock_client.get_sensitive_images.return_value = mock_result
+        mock_get_client.return_value = mock_client
+
+        response = client.get("/moderation/sensitive-images")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["image_ids"] == ["image1", "image2"]
+    assert data["urls"] == ["https://example.com/avatar.jpg"]

moderation/src/admin.rs

···
     pub active_uris: Vec<String>,
 }
 
+/// Request to add a sensitive image.
+#[derive(Debug, Deserialize)]
+pub struct AddSensitiveImageRequest {
+    /// R2 storage ID (for track/album artwork)
+    pub image_id: Option<String>,
+    /// Full URL (for external images like avatars)
+    pub url: Option<String>,
+    /// Why this image was flagged
+    pub reason: Option<String>,
+    /// Admin who flagged it
+    pub flagged_by: Option<String>,
+}
+
+/// Response after adding a sensitive image.
+#[derive(Debug, Serialize)]
+pub struct AddSensitiveImageResponse {
+    pub id: i64,
+    pub message: String,
+}
+
+/// Request to remove a sensitive image.
+#[derive(Debug, Deserialize)]
+pub struct RemoveSensitiveImageRequest {
+    pub id: i64,
+}
+
+/// Response after removing a sensitive image.
+#[derive(Debug, Serialize)]
+pub struct RemoveSensitiveImageResponse {
+    pub removed: bool,
+    pub message: String,
+}
+
 /// List all flagged tracks - returns JSON for API, HTML for htmx.
 pub async fn list_flagged(
     State(state): State<AppState>,
···
     Ok(Json(StoreContextResponse {
         message: format!("context stored for {}", request.uri),
     }))
+}
+
+/// Add a sensitive image entry.
+pub async fn add_sensitive_image(
+    State(state): State<AppState>,
+    Json(request): Json<AddSensitiveImageRequest>,
+) -> Result<Json<AddSensitiveImageResponse>, AppError> {
+    let db = state.db.as_ref().ok_or(AppError::LabelerNotConfigured)?;
+
+    // Validate: at least one of image_id or url must be provided
+    if request.image_id.is_none() && request.url.is_none() {
+        return Err(AppError::BadRequest(
+            "at least one of image_id or url must be provided".to_string(),
+        ));
+    }
+
+    tracing::info!(
+        image_id = ?request.image_id,
+        url = ?request.url,
+        reason = ?request.reason,
+        flagged_by = ?request.flagged_by,
+        "adding sensitive image"
+    );
+
+    let id = db
+        .add_sensitive_image(
+            request.image_id.as_deref(),
+            request.url.as_deref(),
+            request.reason.as_deref(),
+            request.flagged_by.as_deref(),
+        )
+        .await?;
+
+    Ok(Json(AddSensitiveImageResponse {
+        id,
+        message: "sensitive image added".to_string(),
+    }))
+}
+
+/// Remove a sensitive image entry.
+pub async fn remove_sensitive_image(
+    State(state): State<AppState>,
+    Json(request): Json<RemoveSensitiveImageRequest>,
+) -> Result<Json<RemoveSensitiveImageResponse>, AppError> {
+    let db = state.db.as_ref().ok_or(AppError::LabelerNotConfigured)?;
+
+    tracing::info!(id = request.id, "removing sensitive image");
+
+    let removed = db.remove_sensitive_image(request.id).await?;
+
+    let message = if removed {
+        format!("sensitive image {} removed", request.id)
+    } else {
+        format!("sensitive image {} not found", request.id)
+    };
+
+    Ok(Json(RemoveSensitiveImageResponse { removed, message }))
 }
 
 /// Serve the admin UI HTML from static file.

moderation/src/auth.rs

···
     // Static files must be public for admin UI CSS/JS to load
     if path == "/"
         || path == "/health"
+        || path == "/sensitive-images"
         || path == "/admin"
         || path.starts_with("/static/")
         || path.starts_with("/xrpc/com.atproto.label.")

moderation/src/db.rs

···
 
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
-use sqlx::{postgres::PgPoolOptions, PgPool};
+use sqlx::{postgres::PgPoolOptions, FromRow, PgPool};
 
 use crate::admin::FlaggedTrack;
 use crate::labels::Label;
+
+/// Sensitive image record from the database.
+#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
+pub struct SensitiveImageRow {
+    pub id: i64,
+    /// R2 storage ID (for track/album artwork)
+    pub image_id: Option<String>,
+    /// Full URL (for external images like avatars)
+    pub url: Option<String>,
+    /// Why this image was flagged
+    pub reason: Option<String>,
+    /// When the image was flagged
+    pub flagged_at: DateTime<Utc>,
+    /// Admin who flagged it
+    pub flagged_by: Option<String>,
+}
 
 /// Type alias for context row from database query.
 type ContextRow = (
···
             .execute(&self.pool)
             .await?;
         sqlx::query("ALTER TABLE label_context ADD COLUMN IF NOT EXISTS resolution_notes TEXT")
+            .execute(&self.pool)
+            .await?;
+
+        // Sensitive images table for content moderation
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS sensitive_images (
+                id BIGSERIAL PRIMARY KEY,
+                image_id TEXT,
+                url TEXT,
+                reason TEXT,
+                flagged_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+                flagged_by TEXT
+            )
+            "#,
+        )
+        .execute(&self.pool)
+        .await?;
+
+        sqlx::query("CREATE INDEX IF NOT EXISTS idx_sensitive_images_image_id ON sensitive_images(image_id)")
+            .execute(&self.pool)
+            .await?;
+        sqlx::query("CREATE INDEX IF NOT EXISTS idx_sensitive_images_url ON sensitive_images(url)")
             .execute(&self.pool)
             .await?;
 
···
             .collect();
 
         Ok(tracks)
+    }
+
+    // -------------------------------------------------------------------------
+    // Sensitive images
+    // -------------------------------------------------------------------------
+
+    /// Get all sensitive images.
+    pub async fn get_sensitive_images(&self) -> Result<Vec<SensitiveImageRow>, sqlx::Error> {
+        sqlx::query_as::<_, SensitiveImageRow>(
+            "SELECT id, image_id, url, reason, flagged_at, flagged_by FROM sensitive_images ORDER BY flagged_at DESC",
+        )
+        .fetch_all(&self.pool)
+        .await
+    }
+
+    /// Add a sensitive image entry.
+    pub async fn add_sensitive_image(
+        &self,
+        image_id: Option<&str>,
+        url: Option<&str>,
+        reason: Option<&str>,
+        flagged_by: Option<&str>,
+    ) -> Result<i64, sqlx::Error> {
+        sqlx::query_scalar::<_, i64>(
+            r#"
+            INSERT INTO sensitive_images (image_id, url, reason, flagged_by)
+            VALUES ($1, $2, $3, $4)
+            RETURNING id
+            "#,
+        )
+        .bind(image_id)
+        .bind(url)
+        .bind(reason)
+        .bind(flagged_by)
+        .fetch_one(&self.pool)
+        .await
+    }
+
+    /// Remove a sensitive image entry by ID.
+    pub async fn remove_sensitive_image(&self, id: i64) -> Result<bool, sqlx::Error> {
+        let result = sqlx::query("DELETE FROM sensitive_images WHERE id = $1")
+            .bind(id)
+            .execute(&self.pool)
+            .await?;
+        Ok(result.rows_affected() > 0)
     }
 }
 

moderation/src/handlers.rs

···
     pub label: Label,
 }
 
+/// Response for sensitive images endpoint.
+#[derive(Debug, Serialize)]
+pub struct SensitiveImagesResponse {
+    /// R2 image IDs (for track/album artwork)
+    pub image_ids: Vec<String>,
+    /// Full URLs (for external images like avatars)
+    pub urls: Vec<String>,
+}
+
 // --- handlers ---
 
 /// Health check endpoint.
···
     }
 
     Ok(Json(EmitLabelResponse { seq, label }))
+}
+
+/// Get all sensitive images (public endpoint).
+///
+/// Returns image_ids (R2 storage IDs) and urls (full URLs) for all flagged images.
+/// Clients should check both lists when determining if an image is sensitive.
+pub async fn get_sensitive_images(
+    State(state): State<AppState>,
+) -> Result<Json<SensitiveImagesResponse>, AppError> {
+    let db = state.db.as_ref().ok_or(AppError::LabelerNotConfigured)?;
+
+    let images = db.get_sensitive_images().await?;
+
+    let image_ids: Vec<String> = images.iter().filter_map(|i| i.image_id.clone()).collect();
+    let urls: Vec<String> = images.iter().filter_map(|i| i.url.clone()).collect();
+
+    Ok(Json(SensitiveImagesResponse { image_ids, urls }))
 }
 
 #[cfg(test)]

moderation/src/main.rs

···
         .route("/", get(handlers::landing))
         // Health check
         .route("/health", get(handlers::health))
+        // Sensitive images (public)
+        .route("/sensitive-images", get(handlers::get_sensitive_images))
         // AuDD scanning
         .route("/scan", post(audd::scan))
         // Label emission (internal API)
···
         .route("/admin/resolve-htmx", post(admin::resolve_flag_htmx))
         .route("/admin/context", post(admin::store_context))
         .route("/admin/active-labels", post(admin::get_active_labels))
+        .route("/admin/sensitive-images", post(admin::add_sensitive_image))
+        .route(
+            "/admin/sensitive-images/remove",
+            post(admin::remove_sensitive_image),
+        )
         // Static files (CSS, JS for admin UI)
         .nest_service("/static", ServeDir::new("static"))
         // ATProto XRPC endpoints (public)

moderation/src/state.rs

···
     #[error("labeler not configured")]
     LabelerNotConfigured,
 
+    #[error("bad request: {0}")]
+    BadRequest(String),
+
     #[error("label error: {0}")]
     Label(#[from] LabelError),
 
···
             AppError::LabelerNotConfigured => {
                 (StatusCode::SERVICE_UNAVAILABLE, "LabelerNotConfigured")
             }
+            AppError::BadRequest(_) => (StatusCode::BAD_REQUEST, "BadRequest"),
             AppError::Label(_) => (StatusCode::INTERNAL_SERVER_ERROR, "LabelError"),
             AppError::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "DatabaseError"),
             AppError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, "IoError"),

scripts/migrate_sensitive_images.py

···
+#!/usr/bin/env -S uv run --script --quiet --with-editable=backend
+# /// script
+# requires-python = ">=3.12"
+# dependencies = [
+#     "httpx",
+#     "pydantic-settings",
+#     "asyncpg",
+#     "sqlalchemy[asyncio]",
+# ]
+# ///
+"""migrate sensitive images from backend database to moderation service.
+
+this script reads sensitive images from the backend database and creates
+them in the moderation service. after migration, the backend will proxy
+sensitive image requests to the moderation service.
+
+usage:
+    uv run scripts/migrate_sensitive_images.py --env prod --dry-run
+    uv run scripts/migrate_sensitive_images.py --env prod
+
+environment variables (set in .env or export):
+    PROD_DATABASE_URL - production database connection string
+    STAGING_DATABASE_URL - staging database connection string
+    DEV_DATABASE_URL - development database connection string
+    MODERATION_SERVICE_URL - URL of moderation service
+    MODERATION_AUTH_TOKEN - auth token for moderation service
+"""
+
+import argparse
+import asyncio
+import os
+from typing import Literal
+
+import httpx
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+Environment = Literal["dev", "staging", "prod"]
+
+
+class MigrationSettings(BaseSettings):
+    """settings for migration script."""
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    dev_database_url: str = Field(default="", validation_alias="DEV_DATABASE_URL")
+    staging_database_url: str = Field(
+        default="", validation_alias="STAGING_DATABASE_URL"
+    )
+    prod_database_url: str = Field(default="", validation_alias="PROD_DATABASE_URL")
+
+    moderation_service_url: str = Field(
+        default="https://moderation.plyr.fm",
+        validation_alias="MODERATION_SERVICE_URL",
+    )
+    moderation_auth_token: str = Field(
+        default="", validation_alias="MODERATION_AUTH_TOKEN"
+    )
+
+    def get_database_url(self, env: Environment) -> str:
+        """get database URL for environment."""
+        urls = {
+            "dev": self.dev_database_url,
+            "staging": self.staging_database_url,
+            "prod": self.prod_database_url,
+        }
+        url = urls.get(env, "")
+        if not url:
+            raise ValueError(f"no database URL configured for {env}")
+        # ensure asyncpg driver is used
+        if url.startswith("postgresql://"):
+            url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
+        return url
+
+    def get_moderation_url(self, env: Environment) -> str:
+        """get moderation service URL for environment."""
+        if env == "dev":
+            return os.getenv("DEV_MODERATION_URL", "http://localhost:8002")
+        elif env == "staging":
+            return os.getenv("STAGING_MODERATION_URL", "https://moderation-stg.plyr.fm")
+        else:
+            return self.moderation_service_url
+
+
+async def fetch_sensitive_images(db_url: str) -> list[dict]:
+    """fetch all sensitive images from backend database."""
+    engine = create_async_engine(db_url)
+
+    async with engine.begin() as conn:
+        result = await conn.execute(
+            text(
+                """
+                SELECT id, image_id, url, reason, flagged_at, flagged_by
+                FROM sensitive_images
+                ORDER BY id
+                """
+            )
+        )
+        rows = result.fetchall()
+
+    await engine.dispose()
+
+    return [
+        {
+            "id": row[0],
+            "image_id": row[1],
+            "url": row[2],
+            "reason": row[3],
+            "flagged_at": row[4].isoformat() if row[4] else None,
+            "flagged_by": row[5],
+        }
+        for row in rows
+    ]
+
+
+async def migrate_to_moderation_service(
+    images: list[dict],
+    moderation_url: str,
+    auth_token: str,
+    dry_run: bool = False,
+) -> tuple[int, int]:
+    """migrate images to moderation service.
+
+    returns:
+        tuple of (success_count, error_count)
+    """
+    success_count = 0
+    error_count = 0
+
+    headers = {"X-Moderation-Key": auth_token}
+
+    async with httpx.AsyncClient(timeout=30.0) as client:
+        for image in images:
+            payload = {
+                "image_id": image["image_id"],
+                "url": image["url"],
+                "reason": image["reason"],
+                "flagged_by": image["flagged_by"],
+            }
+
+            if dry_run:
+                print(f"  [dry-run] would migrate: {payload}")
+                success_count += 1
+                continue
+
+            try:
+                response = await client.post(
+                    f"{moderation_url}/admin/sensitive-images",
+                    json=payload,
+                    headers=headers,
+                )
+                response.raise_for_status()
+                result = response.json()
+                print(f"  migrated id={image['id']} -> moderation id={result['id']}")
+                success_count += 1
+            except Exception as e:
+                print(f"  ERROR migrating id={image['id']}: {e}")
+                error_count += 1
+
+    return success_count, error_count
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="migrate sensitive images to moderation service"
+    )
+    parser.add_argument(
+        "--env",
+        choices=["dev", "staging", "prod"],
+        required=True,
+        help="environment to migrate",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="show what would be migrated without making changes",
+    )
+    args = parser.parse_args()
+
+    settings = MigrationSettings()
+
+    print(f"migrating sensitive images for {args.env}")
+    if args.dry_run:
+        print("(dry run - no changes will be made)")
+
+    # fetch from backend database
+    db_url = settings.get_database_url(args.env)
+    print("\nfetching from backend database...")
+    images = await fetch_sensitive_images(db_url)
+    print(f"found {len(images)} sensitive images")
+
+    if not images:
+        print("nothing to migrate")
+        return
+
+    # migrate to moderation service
+    moderation_url = settings.get_moderation_url(args.env)
+    print(f"\nmigrating to moderation service at {moderation_url}...")
+
+    if not settings.moderation_auth_token and not args.dry_run:
+        print("ERROR: MODERATION_AUTH_TOKEN not set")
+        return
+
+    success, errors = await migrate_to_moderation_service(
+        images,
+        moderation_url,
+        settings.moderation_auth_token,
+        dry_run=args.dry_run,
+    )
+
+    print(f"\ndone: {success} migrated, {errors} errors")
+
+    if not args.dry_run and errors == 0:
+        print(
+            "\nnext steps:\n"
+            "  1. verify data in moderation service: GET /sensitive-images\n"
+            "  2. update backend to proxy to moderation service\n"
+            "  3. optionally drop sensitive_images table from backend db"
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())