feat: wire up copyright scan integration with moderation service (#382)

* feat: wire up copyright scan integration with moderation service

- add ModerationSettings to config (service URL, auth token, timeout)
- create moderation client module to call /scan endpoint
- hook into upload flow with fire-and-forget background task
- scan runs after successful track commit, doesn't affect upload status
- add 8 tests covering success, timeout, disabled, and error scenarios

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

Co-Authored-By: Claude <[e-mail]>

* fix: add missing logfire[sqlalchemy] dep to copyright scan script

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

Co-Authored-By: Claude <[e-mail]>

* fix: store scan errors as clear results instead of leaving tracks unscanned

when moderation service can't process a file (too short, bad format, etc.),
we now store a clear result with error info in raw_response so the track
isn't stuck in limbo forever.

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

Co-Authored-By: Claude <[e-mail]>

---------

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

backend/src/backend/_internal/moderation.py

···
+"""moderation service client for copyright scanning."""
+
+import logging
+from typing import Any
+
+import httpx
+import logfire
+
+from backend.config import settings
+from backend.models import CopyrightScan
+from backend.utilities.database import db_session
+
+logger = logging.getLogger(__name__)
+
+
+async def scan_track_for_copyright(track_id: int, audio_url: str) -> None:
+    """scan a track for potential copyright matches.
+
+    this runs as a fire-and-forget background task. failures are logged
+    but do not affect the upload flow.
+
+    if the scan fails (e.g., audio too short, unreadable format), we store
+    a "clear" result with the error info so the track isn't stuck unscanned.
+
+    args:
+        track_id: database ID of the track to scan
+        audio_url: public URL of the audio file (R2)
+    """
+    if not settings.moderation.enabled:
+        logger.debug("moderation disabled, skipping copyright scan")
+        return
+
+    if not settings.moderation.auth_token:
+        logger.warning("MODERATION_AUTH_TOKEN not set, skipping copyright scan")
+        return
+
+    with logfire.span(
+        "copyright scan",
+        track_id=track_id,
+        audio_url=audio_url,
+    ):
+        try:
+            result = await _call_moderation_service(audio_url)
+            await _store_scan_result(track_id, result)
+        except Exception as e:
+            logger.warning(
+                "copyright scan failed for track %d: %s - storing as clear",
+                track_id,
+                e,
+            )
+            # store as "clear" with error info so track doesn't stay unscanned
+            # this handles cases like: audio too short, unreadable format, etc.
+            await _store_scan_error(track_id, str(e))
+            # don't re-raise - this is fire-and-forget
+
+
+async def _call_moderation_service(audio_url: str) -> dict[str, Any]:
+    """call the moderation service /scan endpoint.
+
+    args:
+        audio_url: public URL of the audio file
+
+    returns:
+        scan result from moderation service
+
+    raises:
+        httpx.HTTPStatusError: on non-2xx response
+        httpx.TimeoutException: on timeout
+    """
+    async with httpx.AsyncClient(
+        timeout=httpx.Timeout(settings.moderation.timeout_seconds)
+    ) as client:
+        response = await client.post(
+            f"{settings.moderation.service_url}/scan",
+            json={"audio_url": audio_url},
+            headers={"X-Moderation-Key": settings.moderation.auth_token},
+        )
+        response.raise_for_status()
+        return response.json()
+
+
+async def _store_scan_result(track_id: int, result: dict[str, Any]) -> None:
+    """store scan result in the database.
+
+    args:
+        track_id: database ID of the track
+        result: scan result from moderation service
+    """
+    async with db_session() as db:
+        scan = CopyrightScan(
+            track_id=track_id,
+            is_flagged=result.get("is_flagged", False),
+            highest_score=result.get("highest_score", 0),
+            matches=result.get("matches", []),
+            raw_response=result.get("raw_response", {}),
+        )
+        db.add(scan)
+        await db.commit()
+
+        logfire.info(
+            "copyright scan stored",
+            track_id=track_id,
+            is_flagged=scan.is_flagged,
+            highest_score=scan.highest_score,
+            match_count=len(scan.matches),
+        )
+
+
+async def _store_scan_error(track_id: int, error: str) -> None:
+    """store a scan error as a clear result.
+
+    when the moderation service can't process a file (too short, bad format, etc.),
+    we still want to record that we tried so the track isn't stuck in limbo.
+
+    args:
+        track_id: database ID of the track
+        error: error message from the failed scan
+    """
+    async with db_session() as db:
+        scan = CopyrightScan(
+            track_id=track_id,
+            is_flagged=False,
+            highest_score=0,
+            matches=[],
+            raw_response={"error": error, "status": "scan_failed"},
+        )
+        db.add(scan)
+        await db.commit()
+
+        logfire.info(
+            "copyright scan error stored as clear",
+            track_id=track_id,
+            error=error,
+        )

backend/src/backend/api/tracks/uploads.py

···
 from backend._internal.audio import AudioFormat
 from backend._internal.image import ImageFormat
 from backend._internal.jobs import job_service
+from backend._internal.moderation import scan_track_for_copyright
 from backend.config import settings
 from backend.models import Artist, Track
 from backend.models.job import JobStatus, JobType
···
                     except Exception as e:
                         logger.warning(
                             f"failed to send notification for track {track.id}: {e}"
+                        )
+
+                    # kick off copyright scan in background (fire-and-forget)
+                    # this runs independently and doesn't affect the upload result
+                    if r2_url:
+                        # intentionally not storing reference - scan failures are logged
+                        # but shouldn't affect the upload result
+                        asyncio.create_task(  # noqa: RUF006
+                            scan_track_for_copyright(track.id, r2_url)
                         )
 
                     await job_service.update_progress(

backend/src/backend/config.py

···
     )
 
 
+class ModerationSettings(AppSettingsSection):
+    """Moderation service configuration."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="MODERATION_",
+        env_file=".env",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    enabled: bool = Field(
+        default=True,
+        description="Enable copyright scanning on upload",
+    )
+    service_url: str = Field(
+        default="https://plyr-moderation.fly.dev",
+        description="URL of the moderation service",
+    )
+    auth_token: str = Field(
+        default="",
+        description="Auth token for moderation service (X-Moderation-Key header)",
+    )
+    timeout_seconds: int = Field(
+        default=300,
+        description="Timeout for moderation service requests",
+    )
+
+
 class RateLimitSettings(AppSettingsSection):
     """Rate limiting configuration."""
 
···
     notify: NotificationSettings = Field(
         default_factory=NotificationSettings,
         description="Notification settings",
+    )
+    moderation: ModerationSettings = Field(
+        default_factory=ModerationSettings,
+        description="Moderation service settings",
     )
 
 

backend/tests/test_moderation.py

···
+"""tests for copyright moderation integration."""
+
+from unittest.mock import AsyncMock, Mock, patch
+
+import httpx
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend._internal.moderation import (
+    _call_moderation_service,
+    _store_scan_result,
+    scan_track_for_copyright,
+)
+from backend.models import Artist, CopyrightScan, Track
+
+
[e-mail]
+def mock_moderation_response() -> dict:
+    """typical response from moderation service."""
+    return {
+        "matches": [
+            {
+                "artist": "Test Artist",
+                "title": "Test Song",
+                "score": 85,
+                "isrc": "USRC12345678",
+            }
+        ],
+        "is_flagged": True,
+        "highest_score": 85,
+        "raw_response": {"status": "success", "result": []},
+    }
+
+
[e-mail]
+def mock_clear_response() -> dict:
+    """response when no copyright matches found."""
+    return {
+        "matches": [],
+        "is_flagged": False,
+        "highest_score": 0,
+        "raw_response": {"status": "success", "result": None},
+    }
+
+
+async def test_call_moderation_service_success(
+    mock_moderation_response: dict,
+) -> None:
+    """test successful call to moderation service."""
+    # use regular Mock for response since httpx Response methods are sync
+    mock_response = Mock()
+    mock_response.json.return_value = mock_moderation_response
+    mock_response.raise_for_status.return_value = None
+
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
+        mock_post.return_value = mock_response
+
+        with patch("backend._internal.moderation.settings") as mock_settings:
+            mock_settings.moderation.service_url = "https://test.example.com"
+            mock_settings.moderation.auth_token = "test-token"
+            mock_settings.moderation.timeout_seconds = 30
+
+            result = await _call_moderation_service("https://example.com/audio.mp3")
+
+    assert result == mock_moderation_response
+    mock_post.assert_called_once()
+    call_kwargs = mock_post.call_args
+    assert call_kwargs.kwargs["json"] == {"audio_url": "https://example.com/audio.mp3"}
+    assert call_kwargs.kwargs["headers"] == {"X-Moderation-Key": "test-token"}
+
+
+async def test_call_moderation_service_timeout() -> None:
+    """test timeout handling."""
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
+        mock_post.side_effect = httpx.TimeoutException("timeout")
+
+        with patch("backend._internal.moderation.settings") as mock_settings:
+            mock_settings.moderation.service_url = "https://test.example.com"
+            mock_settings.moderation.auth_token = "test-token"
+            mock_settings.moderation.timeout_seconds = 30
+
+            with pytest.raises(httpx.TimeoutException):
+                await _call_moderation_service("https://example.com/audio.mp3")
+
+
+async def test_store_scan_result_flagged(
+    db_session: AsyncSession,
+    mock_moderation_response: dict,
+) -> None:
+    """test storing a flagged scan result."""
+    # create test artist and track
+    artist = Artist(
+        did="did:plc:test123",
+        handle="test.bsky.social",
+        display_name="Test User",
+    )
+    db_session.add(artist)
+    await db_session.commit()
+
+    track = Track(
+        title="Test Track",
+        file_id="test_file_123",
+        file_type="mp3",
+        artist_did=artist.did,
+        r2_url="https://example.com/audio.mp3",
+    )
+    db_session.add(track)
+    await db_session.commit()
+
+    await _store_scan_result(track.id, mock_moderation_response)
+
+    # verify scan was stored
+    result = await db_session.execute(
+        select(CopyrightScan).where(CopyrightScan.track_id == track.id)
+    )
+    scan = result.scalar_one()
+
+    assert scan.is_flagged is True
+    assert scan.highest_score == 85
+    assert len(scan.matches) == 1
+    assert scan.matches[0]["artist"] == "Test Artist"
+
+
+async def test_store_scan_result_clear(
+    db_session: AsyncSession,
+    mock_clear_response: dict,
+) -> None:
+    """test storing a clear (no matches) scan result."""
+    # create test artist and track
+    artist = Artist(
+        did="did:plc:test456",
+        handle="clear.bsky.social",
+        display_name="Clear User",
+    )
+    db_session.add(artist)
+    await db_session.commit()
+
+    track = Track(
+        title="Original Track",
+        file_id="original_file_456",
+        file_type="wav",
+        artist_did=artist.did,
+        r2_url="https://example.com/original.wav",
+    )
+    db_session.add(track)
+    await db_session.commit()
+
+    await _store_scan_result(track.id, mock_clear_response)
+
+    # verify scan was stored
+    result = await db_session.execute(
+        select(CopyrightScan).where(CopyrightScan.track_id == track.id)
+    )
+    scan = result.scalar_one()
+
+    assert scan.is_flagged is False
+    assert scan.highest_score == 0
+    assert scan.matches == []
+
+
+async def test_scan_track_disabled() -> None:
+    """test that scanning is skipped when disabled."""
+    with patch("backend._internal.moderation.settings") as mock_settings:
+        mock_settings.moderation.enabled = False
+
+        with patch(
+            "backend._internal.moderation._call_moderation_service"
+        ) as mock_call:
+            await scan_track_for_copyright(1, "https://example.com/audio.mp3")
+
+            # should not call the service when disabled
+            mock_call.assert_not_called()
+
+
+async def test_scan_track_no_auth_token() -> None:
+    """test that scanning is skipped when auth token not configured."""
+    with patch("backend._internal.moderation.settings") as mock_settings:
+        mock_settings.moderation.enabled = True
+        mock_settings.moderation.auth_token = ""
+
+        with patch(
+            "backend._internal.moderation._call_moderation_service"
+        ) as mock_call:
+            await scan_track_for_copyright(1, "https://example.com/audio.mp3")
+
+            # should not call the service without auth token
+            mock_call.assert_not_called()
+
+
+async def test_scan_track_service_error_stores_as_clear(
+    db_session: AsyncSession,
+) -> None:
+    """test that service errors are stored as clear results."""
+    # create test artist and track
+    artist = Artist(
+        did="did:plc:errortest",
+        handle="errortest.bsky.social",
+        display_name="Error Test User",
+    )
+    db_session.add(artist)
+    await db_session.commit()
+
+    track = Track(
+        title="Error Test Track",
+        file_id="error_test_file",
+        file_type="mp3",
+        artist_did=artist.did,
+        r2_url="https://example.com/short.mp3",
+    )
+    db_session.add(track)
+    await db_session.commit()
+
+    with patch("backend._internal.moderation.settings") as mock_settings:
+        mock_settings.moderation.enabled = True
+        mock_settings.moderation.auth_token = "test-token"
+
+        with patch(
+            "backend._internal.moderation._call_moderation_service",
+            new_callable=AsyncMock,
+        ) as mock_call:
+            mock_call.side_effect = httpx.HTTPStatusError(
+                "502 error",
+                request=AsyncMock(),
+                response=AsyncMock(status_code=502),
+            )
+
+            # should not raise - stores error as clear
+            await scan_track_for_copyright(track.id, "https://example.com/short.mp3")
+
+    # verify scan was stored as clear with error info
+    result = await db_session.execute(
+        select(CopyrightScan).where(CopyrightScan.track_id == track.id)
+    )
+    scan = result.scalar_one()
+
+    assert scan.is_flagged is False
+    assert scan.highest_score == 0
+    assert scan.matches == []
+    assert "error" in scan.raw_response
+    assert scan.raw_response["status"] == "scan_failed"
+
+
+async def test_scan_track_full_flow(
+    db_session: AsyncSession,
+    mock_moderation_response: dict,
+) -> None:
+    """test complete scan flow from track to stored result."""
+    # create test artist and track
+    artist = Artist(
+        did="did:plc:fullflow",
+        handle="fullflow.bsky.social",
+        display_name="Full Flow User",
+    )
+    db_session.add(artist)
+    await db_session.commit()
+
+    track = Track(
+        title="Full Flow Track",
+        file_id="fullflow_file",
+        file_type="flac",
+        artist_did=artist.did,
+        r2_url="https://example.com/fullflow.flac",
+    )
+    db_session.add(track)
+    await db_session.commit()
+
+    with patch("backend._internal.moderation.settings") as mock_settings:
+        mock_settings.moderation.enabled = True
+        mock_settings.moderation.auth_token = "test-token"
+        mock_settings.moderation.service_url = "https://test.example.com"
+        mock_settings.moderation.timeout_seconds = 30
+
+        with patch(
+            "backend._internal.moderation._call_moderation_service",
+            new_callable=AsyncMock,
+        ) as mock_call:
+            mock_call.return_value = mock_moderation_response
+
+            assert track.r2_url is not None
+            await scan_track_for_copyright(track.id, track.r2_url)
+
+    # verify scan was stored (need fresh session query)
+    result = await db_session.execute(
+        select(CopyrightScan).where(CopyrightScan.track_id == track.id)
+    )
+    scan = result.scalar_one()
+
+    assert scan.is_flagged is True
+    assert scan.highest_score == 85

scripts/scan_tracks_copyright.py

···
 #     "pydantic-settings",
 #     "sqlalchemy[asyncio]",
 #     "psycopg[binary]",
+#     "logfire[sqlalchemy]",
 # ]
 # ///
 """scan all tracks for copyright using the moderation service.