audio streaming app
plyr.fm
refactor: extract shared post-creation hooks from upload path (#1068)
move post-creation side effects (notifications, copyright scan, embedding
generation, genre classification, cache invalidation) into a shared
`run_post_track_create_hooks()` function. this enables both the upload
path and the upcoming Jetstream ingest path to share the same behavior.
also moves `invalidate_tracks_discovery_cache` from listing.py to
hooks.py and adds `pds_blob_url` helper to atproto client.
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
authored by
nate nowack
Claude Opus 4.6
and committed by
GitHub
bc6a31aa
b2b47128
+506
-76
+7
-1
backend/src/backend/_internal/atproto/__init__.py
+7
-1
backend/src/backend/_internal/atproto/__init__.py
···
1
1
"""ATProto integration for relay."""
2
2
3
-
from backend._internal.atproto.client import BlobRef, PayloadTooLargeError, upload_blob
3
+
from backend._internal.atproto.client import (
4
+
BlobRef,
5
+
PayloadTooLargeError,
6
+
pds_blob_url,
7
+
upload_blob,
8
+
)
4
9
from backend._internal.atproto.profile import (
5
10
fetch_user_avatar,
6
11
fetch_user_profile,
···
34
39
"fetch_user_profile",
35
40
"get_record_public",
36
41
"normalize_avatar_url",
42
+
"pds_blob_url",
37
43
"sync_atproto_records",
38
44
"update_comment_record",
39
45
"update_list_record",
+5
backend/src/backend/_internal/atproto/client.py
+5
backend/src/backend/_internal/atproto/client.py
···
19
19
logger = logging.getLogger(__name__)
20
20
21
21
22
+
def pds_blob_url(pds_url: str, did: str, cid: str) -> str:
23
+
"""construct a public URL to fetch a blob from a PDS."""
24
+
return f"{pds_url}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}"
25
+
26
+
22
27
class PayloadTooLargeError(Exception):
23
28
"""raised when PDS rejects a blob due to size limits."""
24
29
+6
backend/src/backend/_internal/tasks/__init__.py
+6
backend/src/backend/_internal/tasks/__init__.py
···
20
20
schedule_embedding_generation,
21
21
schedule_genre_classification,
22
22
)
23
+
from backend._internal.tasks.hooks import (
24
+
invalidate_tracks_discovery_cache,
25
+
run_post_track_create_hooks,
26
+
)
23
27
from backend._internal.tasks.pds import (
24
28
pds_create_comment,
25
29
pds_create_like,
···
71
75
"background_tasks",
72
76
"classify_genres",
73
77
"generate_embedding",
78
+
"invalidate_tracks_discovery_cache",
74
79
"move_track_audio",
75
80
"pds_create_comment",
76
81
"pds_create_like",
77
82
"pds_delete_comment",
78
83
"pds_delete_like",
79
84
"pds_update_comment",
85
+
"run_post_track_create_hooks",
80
86
"scan_copyright",
81
87
"schedule_album_list_sync",
82
88
"schedule_atproto_sync",
+133
backend/src/backend/_internal/tasks/hooks.py
+133
backend/src/backend/_internal/tasks/hooks.py
···
1
+
"""shared post-creation hooks for tracks.
2
+
3
+
this is the SINGLE place to add new post-creation behavior.
4
+
both the API upload path and Jetstream ingest call run_post_track_create_hooks().
5
+
"""
6
+
7
+
import logging
8
+
9
+
import logfire
10
+
from redis.exceptions import RedisError
11
+
from sqlalchemy import select
12
+
from sqlalchemy.orm import joinedload
13
+
14
+
from backend._internal.atproto.client import pds_blob_url
15
+
from backend._internal.tasks.copyright import schedule_copyright_scan
16
+
from backend._internal.tasks.ml import (
17
+
schedule_embedding_generation,
18
+
schedule_genre_classification,
19
+
)
20
+
from backend.config import settings
21
+
from backend.models import Artist, Track
22
+
from backend.utilities.database import db_session
23
+
from backend.utilities.redis import get_async_redis_client
24
+
25
+
logger = logging.getLogger(__name__)
26
+
27
+
_DISCOVERY_CACHE_KEY = "plyr:tracks:discovery"
28
+
29
+
30
+
async def invalidate_tracks_discovery_cache() -> None:
31
+
"""delete the anonymous discovery feed cache key."""
32
+
try:
33
+
redis = get_async_redis_client()
34
+
await redis.delete(_DISCOVERY_CACHE_KEY)
35
+
except (RuntimeError, RedisError):
36
+
logger.debug("failed to invalidate discovery cache")
37
+
38
+
39
+
async def resolve_audio_url(track_id: int) -> str | None:
40
+
"""resolve a fetchable audio URL for a track (R2 or PDS blob)."""
41
+
async with db_session() as db:
42
+
result = await db.execute(
43
+
select(
44
+
Track.r2_url,
45
+
Track.audio_storage,
46
+
Track.pds_blob_cid,
47
+
Track.artist_did,
48
+
)
49
+
.where(Track.id == track_id)
50
+
.limit(1)
51
+
)
52
+
row = result.first()
53
+
if not row:
54
+
return None
55
+
56
+
r2_url, audio_storage, pds_blob_cid, artist_did = row
57
+
58
+
if r2_url:
59
+
return r2_url
60
+
61
+
if audio_storage == "pds" and pds_blob_cid:
62
+
artist_result = await db.execute(
63
+
select(Artist.pds_url).where(Artist.did == artist_did).limit(1)
64
+
)
65
+
artist_pds_url = artist_result.scalar_one_or_none()
66
+
if artist_pds_url:
67
+
return pds_blob_url(artist_pds_url, artist_did, pds_blob_cid)
68
+
69
+
return None
70
+
71
+
72
+
async def run_post_track_create_hooks(
73
+
track_id: int,
74
+
*,
75
+
audio_url: str | None = None,
76
+
skip_notification: bool = False,
77
+
skip_copyright: bool = False,
78
+
) -> None:
79
+
"""post-creation side effects for tracks — shared between upload and ingest.
80
+
81
+
this is the SINGLE place to add new post-creation behavior.
82
+
both the API upload path and Jetstream ingest call this function.
83
+
"""
84
+
if audio_url is None:
85
+
audio_url = await resolve_audio_url(track_id)
86
+
87
+
# 1. notification
88
+
if not skip_notification:
89
+
await _send_track_notification(track_id)
90
+
91
+
# 2. copyright scan
92
+
if audio_url and not skip_copyright:
93
+
await schedule_copyright_scan(track_id, audio_url)
94
+
95
+
# 3. CLAP embedding
96
+
if audio_url and settings.modal.enabled and settings.turbopuffer.enabled:
97
+
await schedule_embedding_generation(track_id, audio_url)
98
+
99
+
# 4. genre classification
100
+
if audio_url and settings.replicate.enabled:
101
+
await schedule_genre_classification(track_id, audio_url)
102
+
103
+
# 5. discovery cache invalidation
104
+
await invalidate_tracks_discovery_cache()
105
+
106
+
logfire.info(
107
+
"post-create hooks completed",
108
+
track_id=track_id,
109
+
has_audio_url=audio_url is not None,
110
+
)
111
+
112
+
113
+
async def _send_track_notification(track_id: int) -> None:
114
+
"""send notification for new track, if not already sent."""
115
+
from backend._internal.notifications import notification_service
116
+
117
+
try:
118
+
async with db_session() as db:
119
+
track = await db.scalar(
120
+
select(Track)
121
+
.options(joinedload(Track.artist))
122
+
.where(Track.id == track_id)
123
+
)
124
+
if not track:
125
+
logger.warning(f"track {track_id} not found for notification")
126
+
return
127
+
if track.notification_sent:
128
+
return
129
+
await notification_service.send_track_notification(track)
130
+
track.notification_sent = True
131
+
await db.commit()
132
+
except Exception as e:
133
+
logger.warning(f"failed to send notification for track {track_id}: {e}")
-9
backend/src/backend/api/tracks/listing.py
-9
backend/src/backend/api/tracks/listing.py
···
45
45
logger = logging.getLogger(__name__)
46
46
47
47
48
-
async def invalidate_tracks_discovery_cache() -> None:
49
-
"""delete the anonymous discovery feed cache key."""
50
-
try:
51
-
redis = get_async_redis_client()
52
-
await redis.delete(DISCOVERY_CACHE_KEY)
53
-
except (RuntimeError, RedisError):
54
-
logger.debug("failed to invalidate discovery cache")
55
-
56
-
57
48
if TYPE_CHECKING:
58
49
from backend.storage.r2 import R2Storage
59
50
+1
-1
backend/src/backend/api/tracks/mutations.py
+1
-1
backend/src/backend/api/tracks/mutations.py
···
32
32
schedule_album_list_sync,
33
33
schedule_move_track_audio,
34
34
)
35
+
from backend._internal.tasks.hooks import invalidate_tracks_discovery_cache
35
36
from backend.config import settings
36
37
from backend.models import Artist, Track, TrackTag, get_db
37
38
from backend.schemas import MessageResponse, TrackResponse
38
39
from backend.storage import storage
39
40
from backend.utilities.tags import get_or_create_tag, parse_tags_json
40
41
41
-
from .listing import invalidate_tracks_discovery_cache
42
42
from .metadata_service import (
43
43
apply_album_update,
44
44
resolve_feature_handles,
+17
-50
backend/src/backend/api/tracks/uploads.py
+17
-50
backend/src/backend/api/tracks/uploads.py
···
26
26
from sqlalchemy import select
27
27
from sqlalchemy.exc import IntegrityError
28
28
from sqlalchemy.ext.asyncio import AsyncSession
29
-
from sqlalchemy.orm import joinedload
30
29
31
30
from backend._internal import Session as AuthSession
32
31
from backend._internal import has_flag, require_artist_profile
···
41
40
from backend._internal.clients.transcoder import get_transcoder_client
42
41
from backend._internal.image import ImageFormat
43
42
from backend._internal.jobs import job_service
44
-
from backend._internal.tasks import (
45
-
schedule_album_list_sync,
46
-
schedule_copyright_scan,
47
-
schedule_embedding_generation,
48
-
schedule_genre_classification,
49
-
)
43
+
from backend._internal.tasks import schedule_album_list_sync
44
+
from backend._internal.tasks.hooks import run_post_track_create_hooks
50
45
from backend._internal.thumbnails import generate_and_save
51
46
from backend.config import settings
52
47
from backend.models import Artist, Track, UserPreferences
···
59
54
from backend.utilities.rate_limit import limiter
60
55
from backend.utilities.tags import add_tags_to_track, parse_tags_json
61
56
62
-
from .listing import invalidate_tracks_discovery_cache
63
57
from .router import router
64
58
from .services import get_or_create_album
65
59
···
459
453
)
460
454
461
455
462
-
async def _send_track_notification(db: AsyncSession, track_id: int) -> None:
463
-
"""send notification for new track upload."""
464
-
from backend._internal.notifications import notification_service
465
-
466
-
try:
467
-
track = await db.scalar(
468
-
select(Track).options(joinedload(Track.artist)).where(Track.id == track_id)
469
-
)
470
-
if not track:
471
-
logger.warning(f"track {track_id} not found for notification")
472
-
return
473
-
await notification_service.send_track_notification(track)
474
-
track.notification_sent = True
475
-
await db.commit()
476
-
except Exception as e:
477
-
logger.warning(f"failed to send notification for track {track_id}: {e}")
478
-
479
-
480
456
async def _validate_audio(ctx: UploadContext) -> AudioInfo:
481
457
"""phase 1: validate file type, extract duration, check gating requirements."""
482
458
ext = Path(ctx.filename).suffix.lower()
···
752
728
async def _schedule_post_upload(
753
729
ctx: UploadContext, sr: StorageResult, track: Track
754
730
) -> None:
755
-
"""phase 7: post-upload tasks (tags, notifications, background jobs)."""
731
+
"""phase 7: post-upload tasks (tags, album sync, shared hooks)."""
756
732
async with db_session() as db:
757
733
await add_tags_to_track(db, track.id, ctx.tags, ctx.artist_did)
758
734
759
-
# skip notifications and copyright scan for integration tests on staging
760
-
is_integration_test = (
761
-
settings.observability.environment == "staging"
762
-
and "integration-test" in (ctx.tags or [])
763
-
)
764
-
if not is_integration_test:
765
-
await _send_track_notification(db, track.id)
766
-
if sr.r2_url and not is_integration_test:
767
-
await schedule_copyright_scan(track.id, sr.r2_url)
768
-
769
-
# generate CLAP embedding for vibe search
770
-
if sr.r2_url and settings.modal.enabled and settings.turbopuffer.enabled:
771
-
await schedule_embedding_generation(track.id, sr.r2_url)
772
-
773
-
# classify genres via Replicate
774
-
if sr.r2_url and settings.replicate.enabled:
775
-
await schedule_genre_classification(track.id, sr.r2_url)
776
-
777
-
# sync album list record if track is in an album
735
+
# upload-specific: album list sync
778
736
if track.album_id:
779
737
await schedule_album_list_sync(ctx.auth_session.session_id, track.album_id)
780
738
from backend.api.albums import invalidate_album_cache_by_id
781
739
782
740
async with db_session() as db:
783
741
await invalidate_album_cache_by_id(db, track.album_id)
742
+
743
+
# shared post-creation hooks
744
+
is_integration_test = (
745
+
settings.observability.environment == "staging"
746
+
and "integration-test" in (ctx.tags or [])
747
+
)
748
+
await run_post_track_create_hooks(
749
+
track.id,
750
+
audio_url=sr.r2_url,
751
+
skip_notification=is_integration_test,
752
+
skip_copyright=is_integration_test,
753
+
)
784
754
785
755
786
756
async def _process_upload_background(ctx: UploadContext) -> None:
···
820
790
ctx, audio_info, sr, pds_result, image_id, image_url, thumbnail_url
821
791
)
822
792
823
-
# phase 7: post-upload tasks (tags, notifications, background jobs)
793
+
# phase 7: post-upload tasks (tags, album sync, shared hooks)
824
794
await _schedule_post_upload(ctx, sr, track)
825
-
826
-
# invalidate anonymous discovery feed cache
827
-
await invalidate_tracks_discovery_cache()
828
795
829
796
await job_service.update_progress(
830
797
ctx.upload_id,
+4
-6
backend/tests/api/test_tracks_cache.py
+4
-6
backend/tests/api/test_tracks_cache.py
···
4
4
5
5
from redis.exceptions import ConnectionError as RedisConnectionError
6
6
7
+
from backend._internal.tasks.hooks import invalidate_tracks_discovery_cache
7
8
from backend.api.tracks.constants import DISCOVERY_CACHE_KEY
8
-
from backend.api.tracks.listing import (
9
-
TracksListResponse,
10
-
invalidate_tracks_discovery_cache,
11
-
)
9
+
from backend.api.tracks.listing import TracksListResponse
12
10
13
11
SAMPLE_RESPONSE = TracksListResponse(tracks=[], next_cursor=None, has_more=False)
14
12
···
26
24
mock_redis.delete = AsyncMock()
27
25
28
26
with patch(
29
-
"backend.api.tracks.listing.get_async_redis_client",
27
+
"backend._internal.tasks.hooks.get_async_redis_client",
30
28
return_value=mock_redis,
31
29
):
32
30
await invalidate_tracks_discovery_cache()
···
40
38
mock_redis.delete.side_effect = RedisConnectionError("redis down")
41
39
42
40
with patch(
43
-
"backend.api.tracks.listing.get_async_redis_client",
41
+
"backend._internal.tasks.hooks.get_async_redis_client",
44
42
return_value=mock_redis,
45
43
):
46
44
# should not raise
+4
-9
backend/tests/settings/test_notifications.py
+4
-9
backend/tests/settings/test_notifications.py
···
96
96
with 'Instance is not persistent within this Session' when the track was
97
97
created in a different session (the normal upload flow).
98
98
"""
99
-
from backend.api.tracks.uploads import _send_track_notification
99
+
from backend._internal.tasks.hooks import _send_track_notification
100
100
101
101
artist = Artist(
102
102
did="did:plc:notif_test",
···
120
120
mock_service = AsyncMock()
121
121
mock_service.send_track_notification = AsyncMock(return_value=None)
122
122
123
-
with (
124
-
patch(
125
-
"backend.api.tracks.uploads.notification_service", mock_service, create=True
126
-
),
127
-
patch("backend._internal.notifications.notification_service", mock_service),
128
-
):
129
-
# this used to raise DetachedInstanceError
130
-
await _send_track_notification(db_session, track_id)
123
+
with patch("backend._internal.notifications.notification_service", mock_service):
124
+
# _send_track_notification now opens its own session
125
+
await _send_track_notification(track_id)
131
126
132
127
mock_service.send_track_notification.assert_called_once()
133
128
called_track = mock_service.send_track_notification.call_args[0][0]
+329
backend/tests/test_post_create_hooks.py
+329
backend/tests/test_post_create_hooks.py
···
1
+
"""tests for shared post-creation hooks."""
2
+
3
+
from unittest.mock import AsyncMock, patch
4
+
5
+
from sqlalchemy.ext.asyncio import AsyncSession
6
+
7
+
from backend._internal.atproto.client import pds_blob_url
8
+
from backend._internal.tasks.hooks import resolve_audio_url, run_post_track_create_hooks
9
+
from backend.models import Artist, Track
10
+
11
+
# --- fixtures ---
12
+
13
+
MOCK_SETTINGS_PATH = "backend._internal.tasks.hooks.settings"
14
+
MOCK_COPYRIGHT_PATH = "backend._internal.tasks.hooks.schedule_copyright_scan"
15
+
MOCK_EMBEDDING_PATH = "backend._internal.tasks.hooks.schedule_embedding_generation"
16
+
MOCK_GENRE_PATH = "backend._internal.tasks.hooks.schedule_genre_classification"
17
+
MOCK_NOTIFICATION_PATH = "backend._internal.notifications.notification_service"
18
+
MOCK_REDIS_PATH = "backend._internal.tasks.hooks.get_async_redis_client"
19
+
20
+
21
+
async def _create_artist(db: AsyncSession, pds_url: str | None = None) -> Artist:
22
+
artist = Artist(
23
+
did="did:plc:hooks_test",
24
+
handle="hooks.test.social",
25
+
display_name="Hooks Artist",
26
+
pds_url=pds_url or "https://bsky.social",
27
+
)
28
+
db.add(artist)
29
+
await db.commit()
30
+
return artist
31
+
32
+
33
+
async def _create_track(
34
+
db: AsyncSession,
35
+
artist: Artist,
36
+
*,
37
+
r2_url: str | None = "https://r2.example.com/test.mp3",
38
+
audio_storage: str = "r2",
39
+
pds_blob_cid: str | None = None,
40
+
notification_sent: bool = False,
41
+
) -> Track:
42
+
track = Track(
43
+
title="Test Track",
44
+
file_id="hooks_file_001",
45
+
file_type="mp3",
46
+
artist_did=artist.did,
47
+
r2_url=r2_url,
48
+
audio_storage=audio_storage,
49
+
pds_blob_cid=pds_blob_cid,
50
+
notification_sent=notification_sent,
51
+
)
52
+
db.add(track)
53
+
await db.commit()
54
+
return track
55
+
56
+
57
+
# --- pds_blob_url unit test ---
58
+
59
+
60
+
class TestPdsBlobUrl:
61
+
def test_constructs_url(self) -> None:
62
+
url = pds_blob_url("https://bsky.social", "did:plc:abc", "bafycid")
63
+
assert url == (
64
+
"https://bsky.social/xrpc/com.atproto.sync.getBlob"
65
+
"?did=did:plc:abc&cid=bafycid"
66
+
)
67
+
68
+
69
+
# --- resolve_audio_url tests ---
70
+
71
+
72
+
class TestResolveAudioUrl:
73
+
async def test_r2_url(self, db_session: AsyncSession) -> None:
74
+
artist = await _create_artist(db_session)
75
+
track = await _create_track(db_session, artist)
76
+
url = await resolve_audio_url(track.id)
77
+
assert url == "https://r2.example.com/test.mp3"
78
+
79
+
async def test_pds_blob(self, db_session: AsyncSession) -> None:
80
+
artist = await _create_artist(db_session, pds_url="https://pds.example.com")
81
+
track = await _create_track(
82
+
db_session,
83
+
artist,
84
+
r2_url=None,
85
+
audio_storage="pds",
86
+
pds_blob_cid="bafyblob",
87
+
)
88
+
url = await resolve_audio_url(track.id)
89
+
expected = pds_blob_url("https://pds.example.com", artist.did, "bafyblob")
90
+
assert url == expected
91
+
92
+
async def test_none_when_no_audio(self, db_session: AsyncSession) -> None:
93
+
artist = await _create_artist(db_session)
94
+
track = await _create_track(
95
+
db_session, artist, r2_url=None, audio_storage="pds", pds_blob_cid=None
96
+
)
97
+
url = await resolve_audio_url(track.id)
98
+
assert url is None
99
+
100
+
async def test_none_for_missing_track(self, db_session: AsyncSession) -> None:
101
+
url = await resolve_audio_url(999999)
102
+
assert url is None
103
+
104
+
105
+
# --- run_post_track_create_hooks tests ---
106
+
107
+
108
+
class TestRunPostTrackCreateHooks:
109
+
async def test_schedules_copyright_scan(self, db_session: AsyncSession) -> None:
110
+
artist = await _create_artist(db_session)
111
+
track = await _create_track(db_session, artist)
112
+
113
+
with (
114
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock) as mock_copyright,
115
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock),
116
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock),
117
+
patch(MOCK_NOTIFICATION_PATH),
118
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
119
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
120
+
):
121
+
mock_settings.modal.enabled = False
122
+
mock_settings.turbopuffer.enabled = False
123
+
mock_settings.replicate.enabled = False
124
+
await run_post_track_create_hooks(
125
+
track.id, audio_url="https://r2.example.com/test.mp3"
126
+
)
127
+
128
+
mock_copyright.assert_called_once_with(
129
+
track.id, "https://r2.example.com/test.mp3"
130
+
)
131
+
132
+
async def test_schedules_embedding(self, db_session: AsyncSession) -> None:
133
+
artist = await _create_artist(db_session)
134
+
track = await _create_track(db_session, artist)
135
+
136
+
with (
137
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock),
138
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock) as mock_embed,
139
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock),
140
+
patch(MOCK_NOTIFICATION_PATH),
141
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
142
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
143
+
):
144
+
mock_settings.modal.enabled = True
145
+
mock_settings.turbopuffer.enabled = True
146
+
mock_settings.replicate.enabled = False
147
+
await run_post_track_create_hooks(
148
+
track.id, audio_url="https://r2.example.com/test.mp3"
149
+
)
150
+
151
+
mock_embed.assert_called_once_with(track.id, "https://r2.example.com/test.mp3")
152
+
153
+
async def test_schedules_genre_classification(
154
+
self, db_session: AsyncSession
155
+
) -> None:
156
+
artist = await _create_artist(db_session)
157
+
track = await _create_track(db_session, artist)
158
+
159
+
with (
160
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock),
161
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock),
162
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock) as mock_genre,
163
+
patch(MOCK_NOTIFICATION_PATH),
164
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
165
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
166
+
):
167
+
mock_settings.modal.enabled = False
168
+
mock_settings.turbopuffer.enabled = False
169
+
mock_settings.replicate.enabled = True
170
+
await run_post_track_create_hooks(
171
+
track.id, audio_url="https://r2.example.com/test.mp3"
172
+
)
173
+
174
+
mock_genre.assert_called_once_with(track.id, "https://r2.example.com/test.mp3")
175
+
176
+
async def test_sends_notification(self, db_session: AsyncSession) -> None:
177
+
artist = await _create_artist(db_session)
178
+
track = await _create_track(db_session, artist)
179
+
180
+
mock_service = AsyncMock()
181
+
mock_service.send_track_notification = AsyncMock(return_value=None)
182
+
183
+
with (
184
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock),
185
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock),
186
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock),
187
+
patch(MOCK_NOTIFICATION_PATH, mock_service),
188
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
189
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
190
+
):
191
+
mock_settings.modal.enabled = False
192
+
mock_settings.turbopuffer.enabled = False
193
+
mock_settings.replicate.enabled = False
194
+
await run_post_track_create_hooks(
195
+
track.id, audio_url="https://r2.example.com/test.mp3"
196
+
)
197
+
198
+
mock_service.send_track_notification.assert_called_once()
199
+
200
+
async def test_skips_notification_when_flagged(
201
+
self, db_session: AsyncSession
202
+
) -> None:
203
+
artist = await _create_artist(db_session)
204
+
track = await _create_track(db_session, artist)
205
+
206
+
mock_service = AsyncMock()
207
+
mock_service.send_track_notification = AsyncMock(return_value=None)
208
+
209
+
with (
210
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock),
211
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock),
212
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock),
213
+
patch(MOCK_NOTIFICATION_PATH, mock_service),
214
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
215
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
216
+
):
217
+
mock_settings.modal.enabled = False
218
+
mock_settings.turbopuffer.enabled = False
219
+
mock_settings.replicate.enabled = False
220
+
await run_post_track_create_hooks(
221
+
track.id,
222
+
audio_url="https://r2.example.com/test.mp3",
223
+
skip_notification=True,
224
+
)
225
+
226
+
mock_service.send_track_notification.assert_not_called()
227
+
228
+
async def test_skips_notification_when_already_sent(
229
+
self, db_session: AsyncSession
230
+
) -> None:
231
+
artist = await _create_artist(db_session)
232
+
track = await _create_track(db_session, artist, notification_sent=True)
233
+
234
+
mock_service = AsyncMock()
235
+
mock_service.send_track_notification = AsyncMock(return_value=None)
236
+
237
+
with (
238
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock),
239
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock),
240
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock),
241
+
patch(MOCK_NOTIFICATION_PATH, mock_service),
242
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
243
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
244
+
):
245
+
mock_settings.modal.enabled = False
246
+
mock_settings.turbopuffer.enabled = False
247
+
mock_settings.replicate.enabled = False
248
+
await run_post_track_create_hooks(
249
+
track.id, audio_url="https://r2.example.com/test.mp3"
250
+
)
251
+
252
+
mock_service.send_track_notification.assert_not_called()
253
+
254
+
async def test_skips_copyright_when_flagged(self, db_session: AsyncSession) -> None:
255
+
artist = await _create_artist(db_session)
256
+
track = await _create_track(db_session, artist)
257
+
258
+
with (
259
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock) as mock_copyright,
260
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock),
261
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock),
262
+
patch(MOCK_NOTIFICATION_PATH),
263
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
264
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
265
+
):
266
+
mock_settings.modal.enabled = False
267
+
mock_settings.turbopuffer.enabled = False
268
+
mock_settings.replicate.enabled = False
269
+
await run_post_track_create_hooks(
270
+
track.id,
271
+
audio_url="https://r2.example.com/test.mp3",
272
+
skip_copyright=True,
273
+
)
274
+
275
+
mock_copyright.assert_not_called()
276
+
277
+
async def test_invalidates_cache(self, db_session: AsyncSession) -> None:
278
+
artist = await _create_artist(db_session)
279
+
track = await _create_track(db_session, artist)
280
+
281
+
mock_redis = AsyncMock()
282
+
283
+
with (
284
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock),
285
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock),
286
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock),
287
+
patch(MOCK_NOTIFICATION_PATH),
288
+
patch(MOCK_REDIS_PATH, return_value=mock_redis),
289
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
290
+
):
291
+
mock_settings.modal.enabled = False
292
+
mock_settings.turbopuffer.enabled = False
293
+
mock_settings.replicate.enabled = False
294
+
await run_post_track_create_hooks(
295
+
track.id, audio_url="https://r2.example.com/test.mp3"
296
+
)
297
+
298
+
mock_redis.delete.assert_called_once_with("plyr:tracks:discovery")
299
+
300
+
async def test_no_audio_url_skips_audio_tasks(
301
+
self, db_session: AsyncSession
302
+
) -> None:
303
+
artist = await _create_artist(db_session)
304
+
track = await _create_track(
305
+
db_session, artist, r2_url=None, audio_storage="pds", pds_blob_cid=None
306
+
)
307
+
308
+
mock_service = AsyncMock()
309
+
mock_service.send_track_notification = AsyncMock(return_value=None)
310
+
311
+
with (
312
+
patch(MOCK_COPYRIGHT_PATH, new_callable=AsyncMock) as mock_copyright,
313
+
patch(MOCK_EMBEDDING_PATH, new_callable=AsyncMock) as mock_embed,
314
+
patch(MOCK_GENRE_PATH, new_callable=AsyncMock) as mock_genre,
315
+
patch(MOCK_NOTIFICATION_PATH, mock_service),
316
+
patch(MOCK_REDIS_PATH, return_value=AsyncMock()),
317
+
patch(MOCK_SETTINGS_PATH) as mock_settings,
318
+
):
319
+
mock_settings.modal.enabled = True
320
+
mock_settings.turbopuffer.enabled = True
321
+
mock_settings.replicate.enabled = True
322
+
await run_post_track_create_hooks(track.id, audio_url=None)
323
+
324
+
# audio-dependent tasks not called
325
+
mock_copyright.assert_not_called()
326
+
mock_embed.assert_not_called()
327
+
mock_genre.assert_not_called()
328
+
# notification still fires
329
+
mock_service.send_track_notification.assert_called_once()