audio streaming app
plyr.fm
feat(backend): replace audio on existing track (#1311)
* feat(backend): replace audio on existing track via PUT /tracks/{id}/audio
artists currently delete + re-upload to fix bad audio (logfire shows
darkhart.bsky.social did this 3× in ~65min). that loses likes, comments,
plays, and the track URI. add an endpoint that swaps the audio bytes while
keeping the track's stable identity intact.
orchestration is atomic with rollback:
1. validate + store new audio (R2; transcode if lossless)
2. upload to PDS (best-effort, falls back to r2-only on size limit)
3. PUT updated ATProto record (URI stable, new CID)
4. DB row swap in single tx — file_id, r2_url, atproto_record_cid, duration,
pds_blob_*, audio_storage; clears stale genre_predictions provenance
5. delete old R2 object only on success
6. fire post-replace hooks: invalidate old CopyrightScan rows, re-fire
copyright/embedding/genre tasks; never re-notify followers
7. resync album list record so its strongRef carries the new track CID
if step 3 fails, rollback deletes the just-written R2 file and leaves the
track row untouched.
reuses upload phase helpers (_validate_audio, _store_audio, _upload_to_pds)
so the transcode/PDS-blob/gating logic stays in one place.
intentional non-changes:
- labeler labels on the (URI-stable) track are NOT auto-dismissed — that's
a moderation call left for manual review
- likes/playlists/comments retain stale strongRef CIDs; this is the same
CID-churn behavior that PATCH /tracks/{id} produces today
also fixes two pre-existing test failures uncovered while building this:
- conftest pg_trgm extension only created in xdist template path
- moderation report tests leaked rate-limit budget across the session
Co-Authored-By: Claude Opus 4 (1M context) <noreply@anthropic.com>
* fix(audio-replace): tighten rollback scope + handle gated bucket
addresses two review findings on #1311:
1. **post-commit failures triggered rollback (P1)**. the previous orchestrator
wrapped both pre- and post-commit work in one try block. if any side
effect after `_commit_db_swap` raised (post-replace hooks, album resync,
cache invalidation), the except path would delete the new R2 file even
though the track row + ATProto record were already pointing at it —
leaving production with a 404 for the freshly-replaced audio.
split into two phases: pre-commit may rollback; post-commit failures are
logged and swallowed (the swap stands). each post-commit side effect
gets its own try/log so one failure doesn't skip the others.
2. **gated tracks leaked private-bucket objects (P2)**. `R2Storage.delete()`
only probes the public audio + image buckets, so cleanup and rollback
silently no-op'd on supporter-gated tracks (which live in
`private_audio_bucket_name`).
added `delete_gated()` to `R2Storage` + `StorageProtocol` (mirrors
`delete()`'s refcount guard and key probing, against the private bucket).
`_cleanup_old_files` and `_rollback_new_files` now route based on the
track's `support_gate`. also fixes the same pre-existing leak for
gated tracks deleted via the API today (separate latent bug, but the
primitive is now there).
3. **defensive metadata refresh before publish**. a concurrent PATCH that
landed between `_load_and_authorize` and `_publish_record_update` would
have its title / album / features clobbered by the stale snapshot. now
re-loads the row right before building the new ATProto record.
4. **hoist deferred imports** in audio_replace.py + storage/r2.py per the
project's "no unnecessary deferred imports" rule (CLAUDE.md). the
`backend.api.albums` import doesn't have a real circular dep — i'd
copied the pattern from mutations.py without checking.
new tests:
- post-replace hook failure does NOT roll back the new file
- album list sync failure does NOT roll back the new file
- gated track success path uses `delete_gated` for old file
- gated track rollback uses `delete_gated` for new file
- concurrent PATCH title is reflected in the published ATProto record
full xdist suite: 815 passed (was 810 + 5 new tests).
Co-Authored-By: Claude Opus 4 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4 (1M context) <noreply@anthropic.com>
authored by
nate nowack
Claude Opus 4 (1M context)
and committed by
GitHub
f069e543
a4977bc2
+1846
-7
+59
-2
backend/src/backend/_internal/tasks/hooks.py
+59
-2
backend/src/backend/_internal/tasks/hooks.py
···
8
8
9
9
import logfire
10
10
from redis.exceptions import RedisError
11
-
from sqlalchemy import select
11
+
from sqlalchemy import delete, select
12
12
from sqlalchemy.orm import joinedload
13
13
14
14
from backend._internal.atproto.client import pds_blob_url
···
18
18
schedule_genre_classification,
19
19
)
20
20
from backend.config import settings
21
-
from backend.models import Artist, Track
21
+
from backend.models import Artist, CopyrightScan, Track
22
22
from backend.utilities.database import db_session
23
23
from backend.utilities.redis import get_async_redis_client
24
24
···
107
107
108
108
logfire.info(
109
109
"post-create hooks completed",
110
+
track_id=track_id,
111
+
has_audio_url=audio_url is not None,
112
+
)
113
+
114
+
115
+
async def run_post_track_audio_replace_hooks(
116
+
track_id: int,
117
+
*,
118
+
audio_url: str | None = None,
119
+
) -> None:
120
+
"""post-replace side effects for tracks — fired when audio bytes change.
121
+
122
+
differences from `run_post_track_create_hooks`:
123
+
- never sends a creation notification (followers were notified at upload time)
124
+
- invalidates stale CopyrightScan rows before re-scheduling a fresh scan
125
+
- re-runs CLAP embedding (turbopuffer upsert overwrites by track_id)
126
+
- re-runs genre classification only when the upload originally opted in
127
+
via `extra.auto_tag` (we don't want to clobber manual tags)
128
+
129
+
note: a labeler label attached to the (stable) track URI from the OLD audio
130
+
is NOT auto-dismissed here. that's a moderation decision — left for manual
131
+
review via the admin tooling.
132
+
"""
133
+
if audio_url is None:
134
+
audio_url = await resolve_audio_url(track_id)
135
+
136
+
# 1. invalidate stale copyright scan rows so the new scan replaces them
137
+
async with db_session() as db:
138
+
await db.execute(
139
+
delete(CopyrightScan).where(CopyrightScan.track_id == track_id)
140
+
)
141
+
await db.commit()
142
+
143
+
# 2. re-trigger copyright scan against the new audio bytes
144
+
if audio_url:
145
+
await schedule_copyright_scan(track_id, audio_url)
146
+
147
+
# 3. re-trigger CLAP embedding (turbopuffer upsert overwrites)
148
+
if audio_url and settings.modal.enabled and settings.turbopuffer.enabled:
149
+
await schedule_embedding_generation(track_id, audio_url)
150
+
151
+
# 4. re-trigger genre classification only if the original upload opted in.
152
+
# the classify_genres task tracks `extra.genre_predictions_file_id` so it
153
+
# already knows when a re-classification is needed, but we only want auto-tag
154
+
# to fire again if the user originally requested it.
155
+
if audio_url and settings.replicate.enabled:
156
+
async with db_session() as db:
157
+
track = await db.get(Track, track_id)
158
+
should_reclassify = bool(track and track.extra.get("auto_tag"))
159
+
if should_reclassify:
160
+
await schedule_genre_classification(track_id, audio_url)
161
+
162
+
# 5. discovery cache invalidation (in case "for-you" used stale embedding)
163
+
await invalidate_tracks_discovery_cache()
164
+
165
+
logfire.info(
166
+
"post-replace hooks completed",
110
167
track_id=track_id,
111
168
has_audio_url=audio_url is not None,
112
169
)
+1
backend/src/backend/api/tracks/__init__.py
+1
backend/src/backend/api/tracks/__init__.py
···
14
14
from . import uploads as _uploads # /, /uploads/{upload_id}/progress
15
15
from . import comments as _comments # /{track_id}/comments, /comments/{comment_id}
16
16
from . import mutations as _mutations # /{track_id}, /{track_id}/restore-record
17
+
from . import audio_replace as _audio_replace # /{track_id}/audio
17
18
from . import playback as _playback # /{track_id}, /{track_id}/play
18
19
19
20
__all__ = ["router"]
+651
backend/src/backend/api/tracks/audio_replace.py
+651
backend/src/backend/api/tracks/audio_replace.py
···
1
+
"""replace the audio file backing an existing track.
2
+
3
+
PUT /tracks/{track_id}/audio uploads new audio bytes for an existing track,
4
+
keeping the track's stable identity (id, atproto URI, likes, comments, plays)
5
+
while atomically swapping the underlying file in R2 and on the user's PDS.
6
+
7
+
design notes:
8
+
- the track URI never changes, so likes/playlists/comments keep working.
9
+
- the track record CID does change (new audioUrl/audioBlob/duration). this is
10
+
the same CID-churn behavior as PATCH /tracks/{id} today, which the rest of
11
+
the system already tolerates.
12
+
- a labeler label that was attached to the old audio stays attached to the
13
+
(URI-stable) track. operators must dismiss it manually if the new audio is
14
+
clean. this is intentional — automatically dismissing copyright labels would
15
+
be a moderation hole.
16
+
- the old R2 object is deleted only after the PDS write succeeds; the old PDS
17
+
blob is left to PDS garbage collection.
18
+
"""
19
+
20
+
import contextlib
21
+
import logging
22
+
import tempfile
23
+
from dataclasses import dataclass
24
+
from pathlib import Path
25
+
from typing import Annotated
26
+
from urllib.parse import urljoin
27
+
28
+
import logfire
29
+
from fastapi import (
30
+
BackgroundTasks,
31
+
Depends,
32
+
File,
33
+
HTTPException,
34
+
Request,
35
+
UploadFile,
36
+
)
37
+
from sqlalchemy import select
38
+
from sqlalchemy.orm import selectinload
39
+
40
+
from backend._internal import Session as AuthSession
41
+
from backend._internal import get_session, require_auth
42
+
from backend._internal.atproto.records import (
43
+
build_track_record,
44
+
update_record,
45
+
)
46
+
from backend._internal.audio import AudioFormat
47
+
from backend._internal.jobs import job_service
48
+
from backend._internal.tasks import schedule_album_list_sync
49
+
from backend._internal.tasks.hooks import (
50
+
invalidate_tracks_discovery_cache,
51
+
run_post_track_audio_replace_hooks,
52
+
)
53
+
from backend.api.albums import invalidate_album_cache_by_id
54
+
from backend.api.tracks.uploads import (
55
+
AudioInfo,
56
+
PdsBlobResult,
57
+
StorageResult,
58
+
UploadContext,
59
+
UploadPhaseError,
60
+
UploadStartResponse,
61
+
_store_audio,
62
+
_upload_to_pds,
63
+
_validate_audio,
64
+
)
65
+
from backend.config import settings
66
+
from backend.models import Track
67
+
from backend.models.job import JobStatus, JobType
68
+
from backend.storage import storage
69
+
from backend.utilities.database import db_session
70
+
from backend.utilities.hashing import CHUNK_SIZE
71
+
from backend.utilities.rate_limit import limiter
72
+
73
+
from .router import router
74
+
75
+
logger = logging.getLogger(__name__)
76
+
77
+
78
+
@dataclass
79
+
class TrackAudioState:
80
+
"""snapshot of a track's audio fields, captured before replacement.
81
+
82
+
used for both rebuilding the track ATProto record (preserving non-audio
83
+
metadata like title/album/features/image) and for cleanup of the old R2
84
+
object after the new record publishes successfully.
85
+
"""
86
+
87
+
track_id: int
88
+
artist_did: str
89
+
artist_display_name: str
90
+
atproto_record_uri: str
91
+
92
+
# current audio state (for rollback / cleanup)
93
+
old_file_id: str
94
+
old_file_type: str
95
+
old_original_file_id: str | None
96
+
old_original_file_type: str | None
97
+
98
+
# non-audio metadata (preserved across the rebuild)
99
+
title: str
100
+
album: str | None
101
+
duration: int | None # current duration; replaced with new duration in new record
102
+
features: list[dict]
103
+
image_url: str | None
104
+
description: str | None
105
+
support_gate: dict | None
106
+
107
+
108
+
@dataclass
109
+
class ReplaceContext:
110
+
"""all data needed to process an audio replace in the background."""
111
+
112
+
job_id: str
113
+
auth_session: AuthSession
114
+
track_id: int
115
+
file_path: str
116
+
filename: str
117
+
118
+
119
+
async def _load_and_authorize(
120
+
track_id: int, auth_session: AuthSession
121
+
) -> TrackAudioState:
122
+
"""phase 1: load the track, verify ownership, snapshot current audio state."""
123
+
async with db_session() as db:
124
+
result = await db.execute(
125
+
select(Track)
126
+
.options(selectinload(Track.artist))
127
+
.where(Track.id == track_id)
128
+
)
129
+
track = result.scalar_one_or_none()
130
+
131
+
if not track:
132
+
raise UploadPhaseError("track not found")
133
+
if track.artist_did != auth_session.did:
134
+
raise UploadPhaseError("you can only replace audio on your own tracks")
135
+
if not track.atproto_record_uri:
136
+
raise UploadPhaseError(
137
+
"this track has no ATProto record — restore it before replacing audio"
138
+
)
139
+
140
+
return TrackAudioState(
141
+
track_id=track.id,
142
+
artist_did=track.artist_did,
143
+
artist_display_name=track.artist.display_name,
144
+
atproto_record_uri=track.atproto_record_uri,
145
+
old_file_id=track.file_id,
146
+
old_file_type=track.file_type,
147
+
old_original_file_id=track.original_file_id,
148
+
old_original_file_type=track.original_file_type,
149
+
title=track.title,
150
+
album=track.album,
151
+
duration=track.duration,
152
+
features=list(track.features) if track.features else [],
153
+
image_url=await track.get_image_url(),
154
+
description=track.description,
155
+
support_gate=dict(track.support_gate) if track.support_gate else None,
156
+
)
157
+
158
+
159
+
def _build_upload_context_for_phases(
160
+
ctx: ReplaceContext, state: TrackAudioState
161
+
) -> UploadContext:
162
+
"""build a stub UploadContext so the existing phase helpers work unchanged.
163
+
164
+
only the audio-related fields matter to `_validate_audio`, `_store_audio`,
165
+
`_upload_to_pds`. the rest are filled with sensible defaults that the
166
+
helpers don't read (we never call `_create_records`).
167
+
"""
168
+
return UploadContext(
169
+
upload_id=ctx.job_id,
170
+
auth_session=ctx.auth_session,
171
+
file_path=ctx.file_path,
172
+
filename=ctx.filename,
173
+
title=state.title,
174
+
artist_did=state.artist_did,
175
+
album=state.album,
176
+
album_id=None,
177
+
features_json=None,
178
+
tags=[],
179
+
support_gate=state.support_gate,
180
+
)
181
+
182
+
183
+
def _audio_url_for_record(state: TrackAudioState, sr: StorageResult) -> str:
184
+
"""compute the audioUrl field for the new ATProto record.
185
+
186
+
gated tracks point at the auth-protected backend endpoint; public tracks
187
+
point at the R2 custom domain URL.
188
+
"""
189
+
if state.support_gate is not None:
190
+
backend_url = settings.atproto.redirect_uri.rsplit("/", 2)[0]
191
+
return urljoin(backend_url + "/", f"audio/{sr.file_id}")
192
+
assert sr.r2_url is not None # public tracks always have an r2_url
193
+
return sr.r2_url
194
+
195
+
196
+
async def _publish_record_update(
197
+
ctx: ReplaceContext,
198
+
state: TrackAudioState,
199
+
audio_info: AudioInfo,
200
+
sr: StorageResult,
201
+
pds_result: PdsBlobResult | None,
202
+
) -> str:
203
+
"""phase 5: rebuild the track ATProto record with new audio fields.
204
+
205
+
raises UploadPhaseError on failure. the new R2 file is cleaned up by the
206
+
orchestrator on rollback.
207
+
208
+
returns the new record CID.
209
+
"""
210
+
playable_file_type = (
211
+
sr.playable_format.value
212
+
if sr.playable_format
213
+
else Path(ctx.filename).suffix.lower().lstrip(".")
214
+
)
215
+
216
+
new_record = build_track_record(
217
+
title=state.title,
218
+
artist=state.artist_display_name,
219
+
audio_url=_audio_url_for_record(state, sr),
220
+
file_type=playable_file_type,
221
+
album=state.album,
222
+
duration=audio_info.duration,
223
+
features=state.features or None,
224
+
image_url=state.image_url,
225
+
support_gate=state.support_gate,
226
+
audio_blob=pds_result.blob_ref if pds_result else None,
227
+
description=state.description,
228
+
)
229
+
230
+
try:
231
+
_, new_cid = await update_record(
232
+
auth_session=ctx.auth_session,
233
+
record_uri=state.atproto_record_uri,
234
+
record=new_record,
235
+
)
236
+
except Exception as exc:
237
+
logfire.exception(
238
+
"audio replace: failed to update ATProto record",
239
+
track_id=state.track_id,
240
+
record_uri=state.atproto_record_uri,
241
+
)
242
+
raise UploadPhaseError(f"failed to publish updated record: {exc}") from exc
243
+
244
+
return new_cid
245
+
246
+
247
+
async def _commit_db_swap(
248
+
state: TrackAudioState,
249
+
audio_info: AudioInfo,
250
+
sr: StorageResult,
251
+
pds_result: PdsBlobResult | None,
252
+
new_record_cid: str,
253
+
) -> Track:
254
+
"""phase 6: atomically swap track audio fields in a single transaction.
255
+
256
+
clears the auto_tag flag and stale genre prediction provenance so the
257
+
post-replace hooks make a clean re-classification decision.
258
+
"""
259
+
playable_file_type = (
260
+
sr.playable_format.value
261
+
if sr.playable_format
262
+
else state.old_file_type # transcode shouldn't have run if web-playable; safe fallback
263
+
)
264
+
has_pds_blob = bool(pds_result and pds_result.cid)
265
+
266
+
async with db_session() as db:
267
+
track = await db.get(Track, state.track_id)
268
+
if not track:
269
+
# extremely unlikely race: track deleted between authorize and commit
270
+
raise UploadPhaseError("track was deleted during replace")
271
+
272
+
track.file_id = sr.file_id
273
+
track.file_type = playable_file_type
274
+
track.original_file_id = sr.original_file_id
275
+
track.original_file_type = sr.original_file_type
276
+
track.r2_url = sr.r2_url
277
+
track.audio_storage = "both" if has_pds_blob else "r2"
278
+
track.pds_blob_cid = pds_result.cid if pds_result else None
279
+
track.pds_blob_size = pds_result.size if pds_result else None
280
+
track.atproto_record_cid = new_record_cid
281
+
282
+
# update duration; clear stale genre-prediction provenance so a future
283
+
# re-classification doesn't get short-circuited as "already done".
284
+
extra = dict(track.extra) if track.extra else {}
285
+
if audio_info.duration is not None:
286
+
extra["duration"] = audio_info.duration
287
+
extra.pop("genre_predictions", None)
288
+
extra.pop("genre_predictions_file_id", None)
289
+
track.extra = extra
290
+
291
+
await db.commit()
292
+
await db.refresh(track)
293
+
return track
294
+
295
+
296
+
async def _cleanup_old_files(state: TrackAudioState, sr: StorageResult) -> None:
297
+
"""delete the old R2 audio object(s) after a successful swap.
298
+
299
+
routes to `delete_gated` when the track was supporter-gated (private bucket);
300
+
otherwise uses the public-bucket `delete`. skips deletion when the new
301
+
file_id matches the old one (identical bytes — nothing to clean up), and
302
+
silently swallows already-gone errors.
303
+
304
+
note: the gated/non-gated decision uses the OLD `support_gate` because the
305
+
file we're cleaning up is the OLD audio. if a future endpoint flips gating
306
+
*and* replaces audio in the same call, this needs to take the OLD vs NEW
307
+
bucket destination separately.
308
+
"""
309
+
delete_fn = (
310
+
storage.delete_gated if state.support_gate is not None else storage.delete
311
+
)
312
+
if state.old_file_id != sr.file_id:
313
+
with contextlib.suppress(Exception):
314
+
await delete_fn(state.old_file_id, state.old_file_type)
315
+
# transcode originals always go to the public bucket regardless of gating
316
+
# (gated tracks can't be lossless yet — see _store_audio in uploads.py)
317
+
if state.old_original_file_id and state.old_original_file_id != sr.original_file_id:
318
+
with contextlib.suppress(Exception):
319
+
await storage.delete(
320
+
state.old_original_file_id, state.old_original_file_type
321
+
)
322
+
323
+
324
+
async def _maybe_resync_album_list(track: Track, auth_session: AuthSession) -> None:
325
+
"""if the track is in an album, refresh the album list record so its
326
+
embedded strongRef carries the new track CID."""
327
+
if track.album_id:
328
+
await schedule_album_list_sync(auth_session.session_id, track.album_id)
329
+
async with db_session() as db:
330
+
await invalidate_album_cache_by_id(db, track.album_id)
331
+
332
+
333
+
async def _process_replace_background(ctx: ReplaceContext) -> None:
334
+
"""orchestrate the audio replace pipeline.
335
+
336
+
the pipeline is split into two halves around the DB swap:
337
+
338
+
pre-commit (rollback applies — failure deletes the new R2 file):
339
+
1. load + authorize
340
+
2. validate new bytes
341
+
3. store new bytes (R2 + transcode if needed)
342
+
4. upload to PDS (best-effort)
343
+
5. publish updated ATProto record (PUT)
344
+
6. atomically swap the DB row
345
+
346
+
post-commit (NO rollback — the track is replaced; failures are logged):
347
+
7. delete old R2 file
348
+
8. fire post-replace hooks (rescan, re-embed, re-classify)
349
+
9. resync album list record
350
+
10. invalidate discovery cache
351
+
352
+
the split is critical: once `_commit_db_swap` returns, the track row points
353
+
at the new file and the ATProto record is published. tearing down the new
354
+
R2 object at that point would leave production looking at a broken track.
355
+
"""
356
+
new_file_id_for_rollback: str | None = None
357
+
new_original_file_id_for_rollback: str | None = None
358
+
new_original_file_type_for_rollback: str | None = None
359
+
rollback_gated: bool = False
360
+
361
+
with logfire.span(
362
+
"process audio replace background",
363
+
job_id=ctx.job_id,
364
+
track_id=ctx.track_id,
365
+
filename=ctx.filename,
366
+
):
367
+
# ---- pre-commit (rollback applies) ----
368
+
try:
369
+
await job_service.update_progress(
370
+
ctx.job_id, JobStatus.PROCESSING, "preparing audio replace..."
371
+
)
372
+
373
+
state = await _load_and_authorize(ctx.track_id, ctx.auth_session)
374
+
rollback_gated = state.support_gate is not None
375
+
phase_ctx = _build_upload_context_for_phases(ctx, state)
376
+
377
+
audio_info = await _validate_audio(phase_ctx)
378
+
sr = await _store_audio(phase_ctx, audio_info)
379
+
new_file_id_for_rollback = sr.file_id
380
+
new_original_file_id_for_rollback = sr.original_file_id
381
+
new_original_file_type_for_rollback = sr.original_file_type
382
+
383
+
pds_result = await _upload_to_pds(phase_ctx, audio_info, sr)
384
+
385
+
# the PDS upload may have refreshed the OAuth token in-place; pull
386
+
# the freshest session for the upcoming putRecord call.
387
+
if pds_result and (
388
+
refreshed := await get_session(ctx.auth_session.session_id)
389
+
):
390
+
ctx.auth_session = refreshed
391
+
392
+
# defensive re-load: a concurrent PATCH may have updated title /
393
+
# album / features / image / description / support_gate / unlisted
394
+
# while we were uploading. publish with the freshest non-audio
395
+
# metadata so we don't roll the user's edit back on PDS.
396
+
state = await _refresh_metadata_state(state)
397
+
398
+
new_cid = await _publish_record_update(
399
+
ctx, state, audio_info, sr, pds_result
400
+
)
401
+
402
+
track = await _commit_db_swap(state, audio_info, sr, pds_result, new_cid)
403
+
404
+
except UploadPhaseError as e:
405
+
await _rollback_new_files(
406
+
new_file_id_for_rollback,
407
+
new_original_file_id_for_rollback,
408
+
new_original_file_type_for_rollback,
409
+
gated=rollback_gated,
410
+
)
411
+
await job_service.update_progress(
412
+
ctx.job_id, JobStatus.FAILED, "audio replace failed", error=e.error
413
+
)
414
+
_unlink_temp(ctx.file_path)
415
+
return
416
+
except Exception as e:
417
+
await _rollback_new_files(
418
+
new_file_id_for_rollback,
419
+
new_original_file_id_for_rollback,
420
+
new_original_file_type_for_rollback,
421
+
gated=rollback_gated,
422
+
)
423
+
logger.exception(
424
+
"audio replace job %s failed with unexpected error", ctx.job_id
425
+
)
426
+
await job_service.update_progress(
427
+
ctx.job_id,
428
+
JobStatus.FAILED,
429
+
"audio replace failed",
430
+
error=f"unexpected error: {e!s}",
431
+
)
432
+
_unlink_temp(ctx.file_path)
433
+
return
434
+
435
+
# ---- post-commit (NO rollback — the swap is committed) ----
436
+
# each side effect is best-effort. if any fails we log and keep going;
437
+
# the track is already pointing at the new audio.
438
+
try:
439
+
await _cleanup_old_files(state, sr)
440
+
except Exception:
441
+
logger.exception(
442
+
"audio replace: old-file cleanup failed (track is replaced; "
443
+
"old R2 object may be orphaned)",
444
+
)
445
+
446
+
try:
447
+
await run_post_track_audio_replace_hooks(track.id, audio_url=sr.r2_url)
448
+
except Exception:
449
+
logger.exception(
450
+
"audio replace: post-replace hooks failed (track is replaced; "
451
+
"copyright/embedding/genre re-runs may not have been scheduled)",
452
+
)
453
+
454
+
try:
455
+
await _maybe_resync_album_list(track, ctx.auth_session)
456
+
except Exception:
457
+
logger.exception(
458
+
"audio replace: album list resync failed (track is replaced; "
459
+
"album record's strongRef may carry a stale CID until next edit)",
460
+
)
461
+
462
+
with contextlib.suppress(Exception):
463
+
await invalidate_tracks_discovery_cache()
464
+
465
+
await job_service.update_progress(
466
+
ctx.job_id,
467
+
JobStatus.COMPLETED,
468
+
"audio replaced",
469
+
result={
470
+
"track_id": track.id,
471
+
"atproto_uri": track.atproto_record_uri,
472
+
"atproto_cid": track.atproto_record_cid,
473
+
},
474
+
)
475
+
_unlink_temp(ctx.file_path)
476
+
477
+
478
+
def _unlink_temp(file_path: str) -> None:
479
+
"""remove the temp upload file (suppress all errors)."""
480
+
with contextlib.suppress(Exception):
481
+
Path(file_path).unlink(missing_ok=True)
482
+
483
+
484
+
async def _refresh_metadata_state(state: TrackAudioState) -> TrackAudioState:
485
+
"""reload non-audio fields right before publishing the ATProto record.
486
+
487
+
minimizes the window in which a concurrent `PATCH /tracks/{id}` (title,
488
+
description, image, features, support_gate) gets clobbered by the stale
489
+
metadata we captured at `_load_and_authorize` time.
490
+
"""
491
+
async with db_session() as db:
492
+
result = await db.execute(
493
+
select(Track)
494
+
.options(selectinload(Track.artist))
495
+
.where(Track.id == state.track_id)
496
+
)
497
+
track = result.scalar_one_or_none()
498
+
if not track:
499
+
# row vanished between authorize and publish; let the caller
500
+
# raise the same UploadPhaseError it would have on _commit_db_swap.
501
+
raise UploadPhaseError("track was deleted during replace")
502
+
return TrackAudioState(
503
+
track_id=track.id,
504
+
artist_did=track.artist_did,
505
+
artist_display_name=track.artist.display_name,
506
+
atproto_record_uri=track.atproto_record_uri or state.atproto_record_uri,
507
+
old_file_id=state.old_file_id,
508
+
old_file_type=state.old_file_type,
509
+
old_original_file_id=state.old_original_file_id,
510
+
old_original_file_type=state.old_original_file_type,
511
+
title=track.title,
512
+
album=track.album,
513
+
duration=track.duration,
514
+
features=list(track.features) if track.features else [],
515
+
image_url=await track.get_image_url(),
516
+
description=track.description,
517
+
support_gate=dict(track.support_gate) if track.support_gate else None,
518
+
)
519
+
520
+
521
+
async def _rollback_new_files(
522
+
new_file_id: str | None,
523
+
new_original_file_id: str | None,
524
+
new_original_file_type: str | None,
525
+
*,
526
+
gated: bool,
527
+
) -> None:
528
+
"""delete any new R2 object we wrote before discovering the operation must abort.
529
+
530
+
`gated=True` routes the playable-file delete to the private bucket. transcode
531
+
originals always live in the public bucket (gated tracks can't be lossless),
532
+
so the original delete uses the public path unconditionally.
533
+
"""
534
+
if new_file_id:
535
+
delete_fn = storage.delete_gated if gated else storage.delete
536
+
with contextlib.suppress(Exception):
537
+
await delete_fn(new_file_id)
538
+
if new_original_file_id:
539
+
with contextlib.suppress(Exception):
540
+
await storage.delete(new_original_file_id, new_original_file_type)
541
+
542
+
543
+
# -- HTTP surface ----------------------------------------------------------------
544
+
545
+
546
+
@router.put("/{track_id}/audio")
547
+
@limiter.limit(settings.rate_limit.upload_limit)
548
+
async def replace_track_audio(
549
+
request: Request,
550
+
track_id: int,
551
+
background_tasks: BackgroundTasks,
552
+
auth_session: Annotated[AuthSession, Depends(require_auth)],
553
+
file: UploadFile = File(...),
554
+
) -> UploadStartResponse:
555
+
"""Replace the audio file backing an existing track (owner only).
556
+
557
+
The track id, ATProto URI, likes, comments, plays, tags, and album linkage
558
+
all carry over. Only the audio bytes (and derived fields: duration,
559
+
fingerprint, embedding) are replaced.
560
+
561
+
Returns the same `UploadStartResponse` shape as `POST /tracks/`, so callers
562
+
can reuse the existing `GET /tracks/uploads/{upload_id}/progress` SSE
563
+
endpoint to follow progress.
564
+
"""
565
+
if not file.filename:
566
+
raise HTTPException(status_code=400, detail="no filename provided")
567
+
568
+
ext = Path(file.filename).suffix.lower()
569
+
if not (audio_format := AudioFormat.from_extension(ext)):
570
+
raise HTTPException(
571
+
status_code=400,
572
+
detail=(
573
+
f"unsupported file type: {ext}. "
574
+
f"supported: {AudioFormat.supported_extensions_str()}"
575
+
),
576
+
)
577
+
del audio_format # validated; the background pipeline re-derives it
578
+
579
+
# cheap pre-check so we 404/403 before streaming a multi-MB body to disk
580
+
async with db_session() as db:
581
+
result = await db.execute(
582
+
select(Track.artist_did, Track.atproto_record_uri).where(
583
+
Track.id == track_id
584
+
)
585
+
)
586
+
row = result.first()
587
+
if not row:
588
+
raise HTTPException(status_code=404, detail="track not found")
589
+
artist_did, atproto_uri = row
590
+
if artist_did != auth_session.did:
591
+
raise HTTPException(
592
+
status_code=403,
593
+
detail="you can only replace audio on your own tracks",
594
+
)
595
+
if not atproto_uri:
596
+
raise HTTPException(
597
+
status_code=400,
598
+
detail=(
599
+
"this track has no ATProto record — restore it before "
600
+
"replacing audio"
601
+
),
602
+
)
603
+
604
+
# stream the body to a temp file (constant memory, enforce max size)
605
+
file_path: str | None = None
606
+
try:
607
+
max_size = settings.storage.max_upload_size_mb * 1024 * 1024
608
+
bytes_read = 0
609
+
with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
610
+
file_path = tmp.name
611
+
while chunk := await file.read(CHUNK_SIZE):
612
+
bytes_read += len(chunk)
613
+
if bytes_read > max_size:
614
+
tmp.close()
615
+
Path(file_path).unlink(missing_ok=True)
616
+
raise HTTPException(
617
+
status_code=413,
618
+
detail=(
619
+
f"file too large (max "
620
+
f"{settings.storage.max_upload_size_mb}MB)"
621
+
),
622
+
)
623
+
tmp.write(chunk)
624
+
625
+
job_id = await job_service.create_job(
626
+
JobType.UPLOAD,
627
+
auth_session.did,
628
+
"audio replace queued for processing",
629
+
)
630
+
631
+
background_tasks.add_task(
632
+
_process_replace_background,
633
+
ReplaceContext(
634
+
job_id=job_id,
635
+
auth_session=auth_session,
636
+
track_id=track_id,
637
+
file_path=file_path,
638
+
filename=file.filename,
639
+
),
640
+
)
641
+
except Exception:
642
+
if file_path:
643
+
with contextlib.suppress(Exception):
644
+
Path(file_path).unlink(missing_ok=True)
645
+
raise
646
+
647
+
return UploadStartResponse(
648
+
upload_id=job_id,
649
+
status="pending",
650
+
message="audio replace queued for processing",
651
+
)
+4
backend/src/backend/storage/protocol.py
+4
backend/src/backend/storage/protocol.py
+89
-2
backend/src/backend/storage/r2.py
+89
-2
backend/src/backend/storage/r2.py
···
16
16
from backend._internal.audio import AudioFormat
17
17
from backend._internal.image import ImageFormat
18
18
from backend.config import settings
19
+
from backend.models.track import Track
19
20
from backend.utilities.database import db_session
20
21
from backend.utilities.hashing import hash_file_chunked
21
22
···
362
363
if None, falls back to trying all formats (legacy/images).
363
364
"""
364
365
# check refcount before deleting
365
-
from backend.models.track import Track
366
-
367
366
async with db_session() as db:
368
367
stmt = (
369
368
select(func.count()).select_from(Track).where(Track.file_id == file_id)
···
589
588
590
589
logfire.info("R2 gated upload complete", file_id=file_id, key=key)
591
590
return file_id
591
+
592
+
async def delete_gated(self, file_id: str, file_type: str | None = None) -> bool:
593
+
"""delete a gated audio file from the private R2 bucket.
594
+
595
+
mirrors `delete()` (refcount check then key probe) but targets
596
+
`private_audio_bucket_name`. needed because gated audio lives in a
597
+
separate bucket and the public-bucket-only `delete()` would silently
598
+
no-op on it.
599
+
"""
600
+
if not self.private_audio_bucket_name:
601
+
logfire.warning(
602
+
"delete_gated: R2_PRIVATE_BUCKET not configured, skipping",
603
+
file_id=file_id,
604
+
)
605
+
return False
606
+
607
+
# refcount check — same guard as `delete()` so a duplicate gated row
608
+
# doesn't pull the file out from under another track.
609
+
async with db_session() as db:
610
+
stmt = (
611
+
select(func.count()).select_from(Track).where(Track.file_id == file_id)
612
+
)
613
+
result = await db.execute(stmt)
614
+
refcount = result.scalar_one()
615
+
616
+
if refcount > 1:
617
+
logfire.info(
618
+
"skipping R2 gated delete, file still referenced",
619
+
file_id=file_id,
620
+
refcount=refcount,
621
+
)
622
+
return False
623
+
624
+
async with self._s3_client() as client:
625
+
if file_type:
626
+
audio_format = AudioFormat.from_extension(f".{file_type.lower()}")
627
+
if audio_format:
628
+
key = f"audio/{file_id}{audio_format.extension}"
629
+
try:
630
+
await client.head_object(
631
+
Bucket=self.private_audio_bucket_name, Key=key
632
+
)
633
+
await client.delete_object(
634
+
Bucket=self.private_audio_bucket_name, Key=key
635
+
)
636
+
logfire.info(
637
+
"R2 gated file deleted",
638
+
file_id=file_id,
639
+
key=key,
640
+
bucket=self.private_audio_bucket_name,
641
+
)
642
+
return True
643
+
except client.exceptions.ClientError as e:
644
+
logfire.warning(
645
+
"R2 gated delete failed for known file_type",
646
+
file_id=file_id,
647
+
key=key,
648
+
file_type=file_type,
649
+
error=str(e),
650
+
)
651
+
return False
652
+
653
+
# fallback: try every audio format
654
+
for audio_format in AudioFormat:
655
+
key = f"audio/{file_id}{audio_format.extension}"
656
+
try:
657
+
await client.head_object(
658
+
Bucket=self.private_audio_bucket_name, Key=key
659
+
)
660
+
await client.delete_object(
661
+
Bucket=self.private_audio_bucket_name, Key=key
662
+
)
663
+
logfire.info(
664
+
"R2 gated file deleted (fallback)",
665
+
file_id=file_id,
666
+
key=key,
667
+
bucket=self.private_audio_bucket_name,
668
+
)
669
+
return True
670
+
except client.exceptions.ClientError:
671
+
continue
672
+
673
+
logfire.warning(
674
+
"R2 gated delete: no matching file in private bucket",
675
+
file_id=file_id,
676
+
bucket=self.private_audio_bucket_name,
677
+
)
678
+
return False
592
679
593
680
async def generate_presigned_url(
594
681
self,
backend/tests/api/track_audio_replace/__init__.py
backend/tests/api/track_audio_replace/__init__.py
This is a binary file and will not be displayed.
+213
backend/tests/api/track_audio_replace/_helpers.py
+213
backend/tests/api/track_audio_replace/_helpers.py
···
1
+
"""shared helpers (non-fixture) for the audio-replace test suite.
2
+
3
+
pytest fixtures live in `conftest.py`; everything in this file is plain
4
+
imports — call them as functions.
5
+
"""
6
+
7
+
from __future__ import annotations
8
+
9
+
import contextlib
10
+
from collections.abc import Iterator
11
+
from unittest.mock import AsyncMock, patch
12
+
13
+
from backend._internal import Session
14
+
from backend._internal.audio import AudioFormat
15
+
from backend.api.tracks.audio_replace import ReplaceContext
16
+
from backend.api.tracks.uploads import (
17
+
AudioInfo,
18
+
PdsBlobResult,
19
+
StorageResult,
20
+
)
21
+
from backend.models import Track
22
+
23
+
OWNER_DID = "did:plc:owner"
24
+
OTHER_DID = "did:plc:someone-else"
25
+
TRACK_URI = f"at://{OWNER_DID}/fm.plyr.track/3kabcdefghi23"
26
+
27
+
28
+
class MockSession(Session):
29
+
"""drop-in for require_auth that doesn't touch encryption."""
30
+
31
+
def __init__(self, did: str = OWNER_DID):
32
+
self.did = did
33
+
self.handle = "owner.bsky.social"
34
+
self.session_id = "session-id"
35
+
self.access_token = "tok"
36
+
self.refresh_token = "ref"
37
+
self.oauth_session = {
38
+
"did": did,
39
+
"handle": "owner.bsky.social",
40
+
"pds_url": "https://test.pds",
41
+
"authserver_iss": "https://auth.test",
42
+
"scope": "atproto transition:generic",
43
+
"access_token": "tok",
44
+
"refresh_token": "ref",
45
+
"dpop_private_key_pem": "fake_key",
46
+
"dpop_authserver_nonce": "",
47
+
"dpop_pds_nonce": "",
48
+
}
49
+
50
+
51
+
def make_track(
52
+
*,
53
+
artist_did: str = OWNER_DID,
54
+
file_id: str = "old-file-id",
55
+
file_type: str = "mp3",
56
+
record_uri: str | None = TRACK_URI,
57
+
record_cid: str | None = "bafyOLD",
58
+
duration: int | None = 120,
59
+
album_id: str | None = None,
60
+
support_gate: dict | None = None,
61
+
auto_tag: bool = False,
62
+
) -> Track:
63
+
extra: dict = {}
64
+
if duration is not None:
65
+
extra["duration"] = duration
66
+
if auto_tag:
67
+
extra["auto_tag"] = True
68
+
return Track(
69
+
title="My Song",
70
+
artist_did=artist_did,
71
+
file_id=file_id,
72
+
file_type=file_type,
73
+
original_file_id=None,
74
+
original_file_type=None,
75
+
extra=extra,
76
+
atproto_record_uri=record_uri,
77
+
atproto_record_cid=record_cid,
78
+
r2_url=f"https://audio.example/{file_id}.{file_type}",
79
+
audio_storage="r2",
80
+
pds_blob_cid=None,
81
+
notification_sent=True,
82
+
album_id=album_id,
83
+
support_gate=support_gate,
84
+
)
85
+
86
+
87
+
def replace_ctx(track_id: int = 1) -> ReplaceContext:
88
+
return ReplaceContext(
89
+
job_id="job-1",
90
+
auth_session=MockSession(OWNER_DID),
91
+
track_id=track_id,
92
+
file_path="/tmp/fake-replacement.mp3",
93
+
filename="replacement.mp3",
94
+
)
95
+
96
+
97
+
def audio_info(duration: int = 200, is_gated: bool = False) -> AudioInfo:
98
+
"""default duration differs from `make_track` to verify the swap took."""
99
+
return AudioInfo(format=AudioFormat.MP3, duration=duration, is_gated=is_gated)
100
+
101
+
102
+
def storage_result(
103
+
*,
104
+
file_id: str = "new-file-id",
105
+
original_file_id: str | None = None,
106
+
original_file_type: str | None = None,
107
+
r2_url: str | None = "https://audio.example/new-file-id.mp3",
108
+
) -> StorageResult:
109
+
return StorageResult(
110
+
file_id=file_id,
111
+
original_file_id=original_file_id,
112
+
original_file_type=original_file_type,
113
+
playable_format=AudioFormat.MP3,
114
+
r2_url=r2_url,
115
+
transcode_info=None,
116
+
)
117
+
118
+
119
+
def pds_result(
120
+
*, cid: str | None = "bafyNEWBLOB", size: int | None = 4096
121
+
) -> PdsBlobResult:
122
+
return PdsBlobResult(
123
+
blob_ref={"$type": "blob", "ref": {"$link": cid}, "size": size}
124
+
if cid
125
+
else None,
126
+
cid=cid,
127
+
size=size,
128
+
)
129
+
130
+
131
+
@contextlib.contextmanager
132
+
def patched_replace_pipeline(
133
+
*,
134
+
validate: AudioInfo | None = None,
135
+
store: StorageResult | None = None,
136
+
pds: PdsBlobResult | None = None,
137
+
update_record_return: tuple[str, str] = (TRACK_URI, "bafyNEWREC"),
138
+
update_record_side_effect: object = None,
139
+
storage_delete_side_effect: object = None,
140
+
refreshed_session: Session | None = None,
141
+
) -> Iterator[dict]:
142
+
"""patch every external dependency of `_process_replace_background`.
143
+
144
+
yields a dict of the mocks (`update_record`, `storage_delete`, `post_hooks`,
145
+
`schedule_album_sync`) so tests can assert on them. lets each test focus
146
+
on the one or two things it actually verifies.
147
+
"""
148
+
update_record_kwargs: dict = (
149
+
{"side_effect": update_record_side_effect}
150
+
if update_record_side_effect is not None
151
+
else {"return_value": update_record_return}
152
+
)
153
+
storage_delete_kwargs: dict = (
154
+
{"side_effect": storage_delete_side_effect}
155
+
if storage_delete_side_effect is not None
156
+
else {"return_value": True}
157
+
)
158
+
159
+
with (
160
+
patch(
161
+
"backend.api.tracks.audio_replace._validate_audio",
162
+
AsyncMock(return_value=validate or audio_info()),
163
+
),
164
+
patch(
165
+
"backend.api.tracks.audio_replace._store_audio",
166
+
AsyncMock(return_value=store or storage_result()),
167
+
),
168
+
patch(
169
+
"backend.api.tracks.audio_replace._upload_to_pds",
170
+
AsyncMock(return_value=pds if pds is not None else pds_result()),
171
+
),
172
+
patch(
173
+
"backend.api.tracks.audio_replace.update_record",
174
+
AsyncMock(**update_record_kwargs),
175
+
) as mock_update_record,
176
+
patch(
177
+
"backend.api.tracks.audio_replace.storage.delete",
178
+
AsyncMock(**storage_delete_kwargs),
179
+
) as mock_storage_delete,
180
+
patch(
181
+
"backend.api.tracks.audio_replace.storage.delete_gated",
182
+
AsyncMock(return_value=True),
183
+
) as mock_storage_delete_gated,
184
+
patch(
185
+
"backend.api.tracks.audio_replace.run_post_track_audio_replace_hooks",
186
+
new_callable=AsyncMock,
187
+
) as mock_hooks,
188
+
patch(
189
+
"backend.api.tracks.audio_replace.schedule_album_list_sync",
190
+
new_callable=AsyncMock,
191
+
) as mock_album_sync,
192
+
patch(
193
+
"backend.api.albums.invalidate_album_cache_by_id",
194
+
new_callable=AsyncMock,
195
+
),
196
+
patch(
197
+
"backend.api.tracks.audio_replace.invalidate_tracks_discovery_cache",
198
+
new_callable=AsyncMock,
199
+
),
200
+
patch(
201
+
"backend.api.tracks.audio_replace.get_session",
202
+
AsyncMock(return_value=refreshed_session),
203
+
),
204
+
patch("backend.api.tracks.audio_replace.job_service", AsyncMock()),
205
+
patch("pathlib.Path.unlink"),
206
+
):
207
+
yield {
208
+
"update_record": mock_update_record,
209
+
"storage_delete": mock_storage_delete,
210
+
"storage_delete_gated": mock_storage_delete_gated,
211
+
"post_hooks": mock_hooks,
212
+
"schedule_album_sync": mock_album_sync,
213
+
}
+60
backend/tests/api/track_audio_replace/conftest.py
+60
backend/tests/api/track_audio_replace/conftest.py
···
1
+
"""pytest fixtures shared across the audio-replace test suite.
2
+
3
+
scoped to this directory only — no side effects on other tests.
4
+
"""
5
+
6
+
from __future__ import annotations
7
+
8
+
from collections.abc import Generator
9
+
10
+
import pytest
11
+
from fastapi import FastAPI
12
+
from sqlalchemy.ext.asyncio import AsyncSession
13
+
14
+
from backend._internal import Session, require_auth
15
+
from backend.main import app
16
+
from backend.models import Artist
17
+
18
+
from ._helpers import OTHER_DID, OWNER_DID, MockSession
19
+
20
+
21
+
@pytest.fixture
22
+
async def owner(db_session: AsyncSession) -> Artist:
23
+
artist = Artist(did=OWNER_DID, handle="owner.bsky.social", display_name="Owner")
24
+
db_session.add(artist)
25
+
await db_session.commit()
26
+
await db_session.refresh(artist)
27
+
return artist
28
+
29
+
30
+
@pytest.fixture
31
+
async def other_artist(db_session: AsyncSession) -> Artist:
32
+
artist = Artist(did=OTHER_DID, handle="other.bsky.social", display_name="Other")
33
+
db_session.add(artist)
34
+
await db_session.commit()
35
+
await db_session.refresh(artist)
36
+
return artist
37
+
38
+
39
+
@pytest.fixture
40
+
def test_app_owner() -> Generator[FastAPI, None, None]:
41
+
"""app with require_auth overridden to return the OWNER session."""
42
+
43
+
async def _auth() -> Session:
44
+
return MockSession(OWNER_DID)
45
+
46
+
app.dependency_overrides[require_auth] = _auth
47
+
yield app
48
+
app.dependency_overrides.clear()
49
+
50
+
51
+
@pytest.fixture
52
+
def test_app_other() -> Generator[FastAPI, None, None]:
53
+
"""app with require_auth overridden to return a non-owner session."""
54
+
55
+
async def _auth() -> Session:
56
+
return MockSession(OTHER_DID)
57
+
58
+
app.dependency_overrides[require_auth] = _auth
59
+
yield app
60
+
app.dependency_overrides.clear()
+123
backend/tests/api/track_audio_replace/test_endpoint.py
+123
backend/tests/api/track_audio_replace/test_endpoint.py
···
1
+
"""HTTP-layer tests for `PUT /tracks/{track_id}/audio`.
2
+
3
+
these are the fast, "before the upload even gets queued" checks: auth,
4
+
ownership, format validation, and the basic schedule-and-return contract.
5
+
the actual background pipeline lives in `test_pipeline.py`.
6
+
"""
7
+
8
+
from __future__ import annotations
9
+
10
+
from unittest.mock import AsyncMock, patch
11
+
12
+
from fastapi import FastAPI
13
+
from httpx import ASGITransport, AsyncClient
14
+
from sqlalchemy.ext.asyncio import AsyncSession
15
+
16
+
from backend.models import Artist
17
+
18
+
from ._helpers import make_track
19
+
20
+
21
+
class TestEndpointAuthAndValidation:
22
+
"""fast checks that don't exercise the background pipeline."""
23
+
24
+
async def test_404_when_track_missing(
25
+
self, test_app_owner: FastAPI, owner: Artist
26
+
) -> None:
27
+
async with AsyncClient(
28
+
transport=ASGITransport(app=test_app_owner), base_url="http://test"
29
+
) as client:
30
+
resp = await client.put(
31
+
"/tracks/999999/audio",
32
+
files={"file": ("new.mp3", b"\x00" * 32, "audio/mpeg")},
33
+
)
34
+
assert resp.status_code == 404
35
+
36
+
async def test_403_when_not_owner(
37
+
self,
38
+
test_app_other: FastAPI,
39
+
db_session: AsyncSession,
40
+
owner: Artist,
41
+
other_artist: Artist,
42
+
) -> None:
43
+
track = make_track()
44
+
db_session.add(track)
45
+
await db_session.commit()
46
+
await db_session.refresh(track)
47
+
48
+
async with AsyncClient(
49
+
transport=ASGITransport(app=test_app_other), base_url="http://test"
50
+
) as client:
51
+
resp = await client.put(
52
+
f"/tracks/{track.id}/audio",
53
+
files={"file": ("new.mp3", b"\x00" * 32, "audio/mpeg")},
54
+
)
55
+
assert resp.status_code == 403
56
+
57
+
async def test_400_when_unsupported_format(
58
+
self, test_app_owner: FastAPI, db_session: AsyncSession, owner: Artist
59
+
) -> None:
60
+
track = make_track()
61
+
db_session.add(track)
62
+
await db_session.commit()
63
+
await db_session.refresh(track)
64
+
65
+
async with AsyncClient(
66
+
transport=ASGITransport(app=test_app_owner), base_url="http://test"
67
+
) as client:
68
+
resp = await client.put(
69
+
f"/tracks/{track.id}/audio",
70
+
files={
71
+
"file": (
72
+
"evil.exe",
73
+
b"MZ\x00\x00",
74
+
"application/octet-stream",
75
+
)
76
+
},
77
+
)
78
+
assert resp.status_code == 400
79
+
assert "unsupported" in resp.json()["detail"].lower()
80
+
81
+
async def test_400_when_track_has_no_atproto_record(
82
+
self, test_app_owner: FastAPI, db_session: AsyncSession, owner: Artist
83
+
) -> None:
84
+
"""can't replace audio on a track whose ATProto record was lost."""
85
+
track = make_track(record_uri=None, record_cid=None)
86
+
db_session.add(track)
87
+
await db_session.commit()
88
+
await db_session.refresh(track)
89
+
90
+
async with AsyncClient(
91
+
transport=ASGITransport(app=test_app_owner), base_url="http://test"
92
+
) as client:
93
+
resp = await client.put(
94
+
f"/tracks/{track.id}/audio",
95
+
files={"file": ("new.mp3", b"\x00" * 32, "audio/mpeg")},
96
+
)
97
+
assert resp.status_code == 400
98
+
99
+
async def test_returns_upload_id_and_schedules_background(
100
+
self, test_app_owner: FastAPI, db_session: AsyncSession, owner: Artist
101
+
) -> None:
102
+
"""successful enqueue returns the SSE-pollable upload_id."""
103
+
track = make_track()
104
+
db_session.add(track)
105
+
await db_session.commit()
106
+
await db_session.refresh(track)
107
+
108
+
# patch the background hook so the queued task is a no-op
109
+
with patch(
110
+
"backend.api.tracks.audio_replace._process_replace_background",
111
+
new_callable=AsyncMock,
112
+
):
113
+
async with AsyncClient(
114
+
transport=ASGITransport(app=test_app_owner), base_url="http://test"
115
+
) as client:
116
+
resp = await client.put(
117
+
f"/tracks/{track.id}/audio",
118
+
files={"file": ("new.mp3", b"\x00" * 32, "audio/mpeg")},
119
+
)
120
+
assert resp.status_code == 200
121
+
body = resp.json()
122
+
assert body["status"] == "pending"
123
+
assert body["upload_id"]
+144
backend/tests/api/track_audio_replace/test_hooks.py
+144
backend/tests/api/track_audio_replace/test_hooks.py
···
1
+
"""tests for `run_post_track_audio_replace_hooks`.
2
+
3
+
verifies the hook invalidates stale CopyrightScan rows, re-fires copyright/
4
+
embedding/genre tasks against the new audio bytes, and respects the auto-tag
5
+
opt-in for genre re-classification.
6
+
"""
7
+
8
+
from __future__ import annotations
9
+
10
+
from unittest.mock import AsyncMock, patch
11
+
12
+
from sqlalchemy import select
13
+
from sqlalchemy.ext.asyncio import AsyncSession
14
+
15
+
from backend._internal.tasks.hooks import run_post_track_audio_replace_hooks
16
+
from backend.models import Artist, CopyrightScan
17
+
18
+
from ._helpers import make_track
19
+
20
+
21
+
class TestPostReplaceHooks:
22
+
async def test_invalidates_old_copyright_scan_rows_and_reschedules(
23
+
self, db_session: AsyncSession, owner: Artist
24
+
) -> None:
25
+
track = make_track(file_id="NEW")
26
+
db_session.add(track)
27
+
await db_session.commit()
28
+
await db_session.refresh(track)
29
+
30
+
# an old scan that should be invalidated by the replace hook
31
+
db_session.add(
32
+
CopyrightScan(track_id=track.id, is_flagged=False, highest_score=10)
33
+
)
34
+
await db_session.commit()
35
+
36
+
with (
37
+
patch(
38
+
"backend._internal.tasks.hooks.schedule_copyright_scan",
39
+
new_callable=AsyncMock,
40
+
) as mock_scan,
41
+
patch(
42
+
"backend._internal.tasks.hooks.schedule_embedding_generation",
43
+
new_callable=AsyncMock,
44
+
),
45
+
patch(
46
+
"backend._internal.tasks.hooks.schedule_genre_classification",
47
+
new_callable=AsyncMock,
48
+
),
49
+
patch(
50
+
"backend._internal.tasks.hooks.invalidate_tracks_discovery_cache",
51
+
new_callable=AsyncMock,
52
+
),
53
+
patch("backend._internal.tasks.hooks.settings") as mock_settings,
54
+
):
55
+
mock_settings.modal.enabled = False
56
+
mock_settings.turbopuffer.enabled = False
57
+
mock_settings.replicate.enabled = False
58
+
await run_post_track_audio_replace_hooks(
59
+
track.id, audio_url="https://audio.example/NEW.mp3"
60
+
)
61
+
62
+
# old scan was deleted
63
+
remaining = await db_session.execute(
64
+
select(CopyrightScan).where(CopyrightScan.track_id == track.id)
65
+
)
66
+
assert remaining.scalars().all() == []
67
+
68
+
# new scan was scheduled
69
+
mock_scan.assert_called_once_with(track.id, "https://audio.example/NEW.mp3")
70
+
71
+
async def test_only_reclassifies_genres_when_auto_tag_was_set(
72
+
self, db_session: AsyncSession, owner: Artist
73
+
) -> None:
74
+
"""auto-tag was a per-upload opt-in. don't re-fire it on a manual
75
+
replace if the user never asked for it — they may have hand-tagged."""
76
+
# track WITHOUT auto_tag flag
77
+
track = make_track(file_id="NEW", auto_tag=False)
78
+
db_session.add(track)
79
+
await db_session.commit()
80
+
await db_session.refresh(track)
81
+
82
+
with (
83
+
patch(
84
+
"backend._internal.tasks.hooks.schedule_copyright_scan",
85
+
new_callable=AsyncMock,
86
+
),
87
+
patch(
88
+
"backend._internal.tasks.hooks.schedule_embedding_generation",
89
+
new_callable=AsyncMock,
90
+
),
91
+
patch(
92
+
"backend._internal.tasks.hooks.schedule_genre_classification",
93
+
new_callable=AsyncMock,
94
+
) as mock_genre,
95
+
patch(
96
+
"backend._internal.tasks.hooks.invalidate_tracks_discovery_cache",
97
+
new_callable=AsyncMock,
98
+
),
99
+
patch("backend._internal.tasks.hooks.settings") as mock_settings,
100
+
):
101
+
mock_settings.modal.enabled = False
102
+
mock_settings.turbopuffer.enabled = False
103
+
mock_settings.replicate.enabled = True
104
+
await run_post_track_audio_replace_hooks(
105
+
track.id, audio_url="https://audio.example/NEW.mp3"
106
+
)
107
+
108
+
mock_genre.assert_not_called()
109
+
110
+
async def test_reclassifies_genres_when_auto_tag_present(
111
+
self, db_session: AsyncSession, owner: Artist
112
+
) -> None:
113
+
track = make_track(file_id="NEW", auto_tag=True)
114
+
db_session.add(track)
115
+
await db_session.commit()
116
+
await db_session.refresh(track)
117
+
118
+
with (
119
+
patch(
120
+
"backend._internal.tasks.hooks.schedule_copyright_scan",
121
+
new_callable=AsyncMock,
122
+
),
123
+
patch(
124
+
"backend._internal.tasks.hooks.schedule_embedding_generation",
125
+
new_callable=AsyncMock,
126
+
),
127
+
patch(
128
+
"backend._internal.tasks.hooks.schedule_genre_classification",
129
+
new_callable=AsyncMock,
130
+
) as mock_genre,
131
+
patch(
132
+
"backend._internal.tasks.hooks.invalidate_tracks_discovery_cache",
133
+
new_callable=AsyncMock,
134
+
),
135
+
patch("backend._internal.tasks.hooks.settings") as mock_settings,
136
+
):
137
+
mock_settings.modal.enabled = False
138
+
mock_settings.turbopuffer.enabled = False
139
+
mock_settings.replicate.enabled = True
140
+
await run_post_track_audio_replace_hooks(
141
+
track.id, audio_url="https://audio.example/NEW.mp3"
142
+
)
143
+
144
+
mock_genre.assert_called_once_with(track.id, "https://audio.example/NEW.mp3")
+470
backend/tests/api/track_audio_replace/test_pipeline.py
+470
backend/tests/api/track_audio_replace/test_pipeline.py
···
1
+
"""orchestration tests for `_process_replace_background`.
2
+
3
+
these run the background pipeline end-to-end with the I/O phases mocked,
4
+
verifying the swap is atomic, rollback works, and the right hooks fire.
5
+
"""
6
+
7
+
from __future__ import annotations
8
+
9
+
from unittest.mock import AsyncMock, patch
10
+
11
+
from sqlalchemy import update
12
+
from sqlalchemy.ext.asyncio import AsyncSession
13
+
14
+
from backend.api.tracks.audio_replace import (
15
+
TrackAudioState,
16
+
_process_replace_background,
17
+
)
18
+
from backend.models import Album, Artist, Track
19
+
from backend.utilities.database import db_session as _db_session
20
+
21
+
from ._helpers import (
22
+
OWNER_DID,
23
+
TRACK_URI,
24
+
MockSession,
25
+
audio_info,
26
+
make_track,
27
+
patched_replace_pipeline,
28
+
pds_result,
29
+
replace_ctx,
30
+
storage_result,
31
+
)
32
+
33
+
34
+
class TestReplaceOrchestration:
35
+
"""exercise `_process_replace_background` end-to-end with mocked phases."""
36
+
37
+
async def test_happy_path_swaps_audio_and_updates_record(
38
+
self, db_session: AsyncSession, owner: Artist
39
+
) -> None:
40
+
"""on success: file_id/r2_url/atproto_record_cid/duration update,
41
+
old R2 file is deleted, post-replace hooks fire, no notification fires."""
42
+
track = make_track(file_id="OLD", duration=120)
43
+
db_session.add(track)
44
+
await db_session.commit()
45
+
await db_session.refresh(track)
46
+
track_id = track.id
47
+
48
+
deleted_keys: list[str] = []
49
+
50
+
async def fake_delete(file_id: str, file_type: str | None = None) -> bool:
51
+
deleted_keys.append(file_id)
52
+
return True
53
+
54
+
with patched_replace_pipeline(
55
+
store=storage_result(file_id="NEW"),
56
+
storage_delete_side_effect=fake_delete,
57
+
) as mocks:
58
+
await _process_replace_background(replace_ctx(track_id=track_id))
59
+
60
+
# DB row was atomically swapped
61
+
await db_session.refresh(track)
62
+
assert track.file_id == "NEW"
63
+
assert track.r2_url == "https://audio.example/new-file-id.mp3"
64
+
assert track.atproto_record_cid == "bafyNEWREC"
65
+
assert track.atproto_record_uri == TRACK_URI # URI is stable
66
+
assert track.extra["duration"] == 200 # new duration
67
+
assert track.audio_storage == "both"
68
+
assert track.pds_blob_cid == "bafyNEWBLOB"
69
+
assert track.notification_sent is True # never re-fires
70
+
71
+
# old R2 file was deleted
72
+
assert "OLD" in deleted_keys
73
+
74
+
# post-replace hooks were scheduled with the new audio URL
75
+
mocks["post_hooks"].assert_called_once()
76
+
call = mocks["post_hooks"].call_args
77
+
assert call.args[0] == track_id
78
+
assert call.kwargs["audio_url"] == "https://audio.example/new-file-id.mp3"
79
+
80
+
async def test_rollback_when_pds_record_update_fails(
81
+
self, db_session: AsyncSession, owner: Artist
82
+
) -> None:
83
+
"""if `update_record` raises, we delete the new R2 file and leave the
84
+
track row pointing at the OLD file_id/cid — no orphan, no broken state."""
85
+
track = make_track(file_id="OLD", record_cid="bafyOLD", duration=120)
86
+
db_session.add(track)
87
+
await db_session.commit()
88
+
await db_session.refresh(track)
89
+
track_id = track.id
90
+
91
+
deleted_keys: list[str] = []
92
+
93
+
async def fake_delete(file_id: str, file_type: str | None = None) -> bool:
94
+
deleted_keys.append(file_id)
95
+
return True
96
+
97
+
with patched_replace_pipeline(
98
+
store=storage_result(file_id="NEW"),
99
+
update_record_side_effect=RuntimeError("PDS exploded"),
100
+
storage_delete_side_effect=fake_delete,
101
+
) as mocks:
102
+
await _process_replace_background(replace_ctx(track_id=track_id))
103
+
104
+
# track row is unchanged
105
+
await db_session.refresh(track)
106
+
assert track.file_id == "OLD"
107
+
assert track.atproto_record_cid == "bafyOLD"
108
+
assert track.extra["duration"] == 120 # never updated
109
+
110
+
# the new R2 file we just wrote was deleted (no orphan)
111
+
assert "NEW" in deleted_keys
112
+
# the old R2 file was NOT deleted
113
+
assert "OLD" not in deleted_keys
114
+
115
+
# post-replace hooks must not have fired
116
+
mocks["post_hooks"].assert_not_called()
117
+
118
+
async def test_pds_blob_too_large_fallback_keeps_r2_only(
119
+
self, db_session: AsyncSession, owner: Artist
120
+
) -> None:
121
+
"""when PDS upload returns a None-cid result (too large / failed), the
122
+
track is recorded as r2-only and pds_blob_cid is cleared."""
123
+
track = make_track(file_id="OLD")
124
+
# simulate that the previous audio had a PDS blob
125
+
track.pds_blob_cid = "bafyOLDBLOB"
126
+
track.pds_blob_size = 9999
127
+
track.audio_storage = "both"
128
+
db_session.add(track)
129
+
await db_session.commit()
130
+
await db_session.refresh(track)
131
+
track_id = track.id
132
+
133
+
with patched_replace_pipeline(
134
+
store=storage_result(file_id="NEW"),
135
+
pds=pds_result(cid=None, size=None),
136
+
):
137
+
await _process_replace_background(replace_ctx(track_id=track_id))
138
+
139
+
await db_session.refresh(track)
140
+
assert track.audio_storage == "r2"
141
+
assert track.pds_blob_cid is None
142
+
assert track.pds_blob_size is None
143
+
144
+
async def test_album_list_resync_scheduled_when_track_in_album(
145
+
self, db_session: AsyncSession, owner: Artist
146
+
) -> None:
147
+
"""tracks in an album must trigger a list-record resync so the album's
148
+
embedded strongRef carries the new track CID."""
149
+
album = Album(artist_did=OWNER_DID, slug="my-album", title="My Album")
150
+
db_session.add(album)
151
+
await db_session.flush()
152
+
track = make_track(album_id=album.id)
153
+
db_session.add(track)
154
+
await db_session.commit()
155
+
await db_session.refresh(track)
156
+
track_id = track.id
157
+
album_id = album.id
158
+
159
+
with patched_replace_pipeline(
160
+
store=storage_result(file_id="NEW"),
161
+
) as mocks:
162
+
await _process_replace_background(replace_ctx(track_id=track_id))
163
+
164
+
mocks["schedule_album_sync"].assert_called_once()
165
+
# second positional arg is album_id (after session_id)
166
+
assert mocks["schedule_album_sync"].call_args.args[1] == album_id
167
+
168
+
async def test_gated_track_record_uses_backend_audio_url(
169
+
self, db_session: AsyncSession, owner: Artist
170
+
) -> None:
171
+
"""gated tracks must not write a public R2 URL into the ATProto record;
172
+
they need the auth-protected `/audio/{file_id}` redirect endpoint."""
173
+
track = make_track(support_gate={"type": "any"})
174
+
db_session.add(track)
175
+
await db_session.commit()
176
+
await db_session.refresh(track)
177
+
track_id = track.id
178
+
179
+
captured_record: dict = {}
180
+
181
+
async def fake_update_record(
182
+
*, auth_session, record_uri: str, record: dict
183
+
) -> tuple[str, str]:
184
+
captured_record.update(record)
185
+
return record_uri, "bafyNEWREC"
186
+
187
+
with patched_replace_pipeline(
188
+
validate=audio_info(is_gated=True),
189
+
store=storage_result(file_id="NEW", r2_url=None), # gated → r2_url=None
190
+
pds=None, # gated tracks skip PDS upload
191
+
update_record_side_effect=fake_update_record,
192
+
):
193
+
await _process_replace_background(replace_ctx(track_id=track_id))
194
+
195
+
# the recorded audioUrl points at /audio/{file_id}, not at R2
196
+
assert captured_record["audioUrl"].endswith("/audio/NEW")
197
+
assert "supportGate" in captured_record
198
+
assert captured_record["supportGate"] == {"type": "any"}
199
+
200
+
async def test_lossless_replace_records_original_file_id(
201
+
self, db_session: AsyncSession, owner: Artist
202
+
) -> None:
203
+
"""a lossless (FLAC) replacement should store both transcoded mp3 and
204
+
the original file id on the row."""
205
+
track = make_track(file_id="OLD")
206
+
db_session.add(track)
207
+
await db_session.commit()
208
+
await db_session.refresh(track)
209
+
track_id = track.id
210
+
211
+
with patched_replace_pipeline(
212
+
store=storage_result(
213
+
file_id="NEW_MP3",
214
+
original_file_id="NEW_FLAC",
215
+
original_file_type="flac",
216
+
),
217
+
):
218
+
await _process_replace_background(replace_ctx(track_id=track_id))
219
+
220
+
await db_session.refresh(track)
221
+
assert track.file_id == "NEW_MP3"
222
+
assert track.original_file_id == "NEW_FLAC"
223
+
assert track.original_file_type == "flac"
224
+
225
+
async def test_db_swap_clears_stale_genre_predictions(
226
+
self, db_session: AsyncSession, owner: Artist
227
+
) -> None:
228
+
"""stale genre_predictions / genre_predictions_file_id must be cleared
229
+
on a successful audio swap so a future re-classification doesn't get
230
+
short-circuited as 'already done'."""
231
+
track = make_track(file_id="OLD")
232
+
track.extra = {
233
+
**track.extra,
234
+
"genre_predictions": [{"name": "ambient", "confidence": 0.9}],
235
+
"genre_predictions_file_id": "OLD",
236
+
}
237
+
db_session.add(track)
238
+
await db_session.commit()
239
+
await db_session.refresh(track)
240
+
track_id = track.id
241
+
242
+
with patched_replace_pipeline(store=storage_result(file_id="NEW")):
243
+
await _process_replace_background(replace_ctx(track_id=track_id))
244
+
245
+
await db_session.refresh(track)
246
+
assert "genre_predictions" not in track.extra
247
+
assert "genre_predictions_file_id" not in track.extra
248
+
249
+
async def test_session_reloaded_after_pds_upload(
250
+
self, db_session: AsyncSession, owner: Artist
251
+
) -> None:
252
+
"""if PDS upload refreshed the OAuth token, the new session is used for
253
+
the subsequent putRecord call (mirrors the upload-pipeline regression)."""
254
+
track = make_track(file_id="OLD")
255
+
db_session.add(track)
256
+
await db_session.commit()
257
+
await db_session.refresh(track)
258
+
track_id = track.id
259
+
260
+
refreshed = MockSession(OWNER_DID)
261
+
refreshed.oauth_session["access_token"] = "REFRESHED-TOKEN"
262
+
263
+
captured_session: dict = {}
264
+
265
+
async def fake_update_record(
266
+
*, auth_session, record_uri: str, record: dict
267
+
) -> tuple[str, str]:
268
+
captured_session["token"] = auth_session.oauth_session["access_token"]
269
+
return record_uri, "bafyNEWREC"
270
+
271
+
with patched_replace_pipeline(
272
+
store=storage_result(file_id="NEW"),
273
+
update_record_side_effect=fake_update_record,
274
+
refreshed_session=refreshed,
275
+
):
276
+
await _process_replace_background(replace_ctx(track_id=track_id))
277
+
278
+
assert captured_session["token"] == "REFRESHED-TOKEN"
279
+
280
+
281
+
class TestPostCommitFailureIsolation:
282
+
"""regression for review feedback: post-commit failures must NOT roll back
283
+
the new audio. once `_commit_db_swap` returns, the track is replaced —
284
+
deleting the new R2 file at that point would leave production broken."""
285
+
286
+
async def test_post_replace_hooks_failure_does_not_rollback(
287
+
self, db_session: AsyncSession, owner: Artist
288
+
) -> None:
289
+
track = make_track(file_id="OLD")
290
+
db_session.add(track)
291
+
await db_session.commit()
292
+
await db_session.refresh(track)
293
+
track_id = track.id
294
+
295
+
deleted_keys: list[str] = []
296
+
297
+
async def fake_delete(file_id: str, file_type: str | None = None) -> bool:
298
+
deleted_keys.append(file_id)
299
+
return True
300
+
301
+
with patched_replace_pipeline(
302
+
store=storage_result(file_id="NEW"),
303
+
storage_delete_side_effect=fake_delete,
304
+
) as mocks:
305
+
# post-commit hook blows up
306
+
mocks["post_hooks"].side_effect = RuntimeError("docket exploded")
307
+
await _process_replace_background(replace_ctx(track_id=track_id))
308
+
309
+
# track row IS replaced — no rollback
310
+
await db_session.refresh(track)
311
+
assert track.file_id == "NEW"
312
+
assert track.atproto_record_cid == "bafyNEWREC"
313
+
314
+
# the NEW R2 file must NOT have been deleted
315
+
assert "NEW" not in deleted_keys
316
+
# the OLD R2 file should have been cleaned up before the hooks ran
317
+
assert "OLD" in deleted_keys
318
+
319
+
async def test_album_resync_failure_does_not_rollback(
320
+
self, db_session: AsyncSession, owner: Artist
321
+
) -> None:
322
+
album = Album(artist_did=OWNER_DID, slug="my-album", title="My Album")
323
+
db_session.add(album)
324
+
await db_session.flush()
325
+
track = make_track(album_id=album.id, file_id="OLD")
326
+
db_session.add(track)
327
+
await db_session.commit()
328
+
await db_session.refresh(track)
329
+
track_id = track.id
330
+
331
+
with patched_replace_pipeline(
332
+
store=storage_result(file_id="NEW"),
333
+
) as mocks:
334
+
mocks["schedule_album_sync"].side_effect = RuntimeError("docket down")
335
+
await _process_replace_background(replace_ctx(track_id=track_id))
336
+
337
+
# row is committed despite the schedule_album_list_sync failure
338
+
await db_session.refresh(track)
339
+
assert track.file_id == "NEW"
340
+
assert track.atproto_record_cid == "bafyNEWREC"
341
+
342
+
343
+
class TestGatedAudioCleanup:
344
+
"""regression for review feedback: gated tracks live in the private R2
345
+
bucket. cleanup and rollback must use `delete_gated`, not `delete`."""
346
+
347
+
async def test_old_gated_audio_uses_delete_gated_on_success(
348
+
self, db_session: AsyncSession, owner: Artist
349
+
) -> None:
350
+
track = make_track(file_id="OLD", support_gate={"type": "any"})
351
+
db_session.add(track)
352
+
await db_session.commit()
353
+
await db_session.refresh(track)
354
+
track_id = track.id
355
+
356
+
with patched_replace_pipeline(
357
+
validate=audio_info(is_gated=True),
358
+
store=storage_result(file_id="NEW", r2_url=None),
359
+
pds=None, # gated tracks skip PDS upload
360
+
) as mocks:
361
+
await _process_replace_background(replace_ctx(track_id=track_id))
362
+
363
+
# the OLD gated file was deleted via delete_gated, NOT delete
364
+
mocks["storage_delete_gated"].assert_called_once()
365
+
assert mocks["storage_delete_gated"].call_args.args[0] == "OLD"
366
+
# delete (public bucket) must not have been used for the gated file_id
367
+
for call in mocks["storage_delete"].call_args_list:
368
+
assert call.args[0] != "OLD"
369
+
370
+
async def test_rollback_for_gated_track_uses_delete_gated(
371
+
self, db_session: AsyncSession, owner: Artist
372
+
) -> None:
373
+
track = make_track(file_id="OLD", support_gate={"type": "any"})
374
+
db_session.add(track)
375
+
await db_session.commit()
376
+
await db_session.refresh(track)
377
+
track_id = track.id
378
+
379
+
with patched_replace_pipeline(
380
+
validate=audio_info(is_gated=True),
381
+
store=storage_result(file_id="NEW", r2_url=None),
382
+
pds=None,
383
+
update_record_side_effect=RuntimeError("PDS exploded"),
384
+
) as mocks:
385
+
await _process_replace_background(replace_ctx(track_id=track_id))
386
+
387
+
# rollback deletes the NEW file from the PRIVATE bucket
388
+
mocks["storage_delete_gated"].assert_called_once()
389
+
assert mocks["storage_delete_gated"].call_args.args[0] == "NEW"
390
+
391
+
# track row is unchanged
392
+
await db_session.refresh(track)
393
+
assert track.file_id == "OLD"
394
+
395
+
396
+
class TestConcurrentMetadataPatch:
397
+
"""regression: minimize the window where a concurrent PATCH /tracks/{id}
398
+
title (or other metadata) gets clobbered by a stale snapshot taken at
399
+
`_load_and_authorize` time."""
400
+
401
+
async def test_publishes_freshly_loaded_title(
402
+
self, db_session: AsyncSession, owner: Artist
403
+
) -> None:
404
+
track = make_track(file_id="OLD")
405
+
track.title = "Original Title"
406
+
db_session.add(track)
407
+
await db_session.commit()
408
+
await db_session.refresh(track)
409
+
track_id = track.id
410
+
411
+
captured_record: dict = {}
412
+
413
+
async def fake_update_record(
414
+
*, auth_session, record_uri: str, record: dict
415
+
) -> tuple[str, str]:
416
+
captured_record.update(record)
417
+
return record_uri, "bafyNEWREC"
418
+
419
+
# simulate a concurrent PATCH that lands AFTER _load_and_authorize but
420
+
# BEFORE _publish_record_update by mutating the row mid-pipeline — the
421
+
# mutation runs in the side_effect of `_store_audio`, which fires
422
+
# between authorize and publish.
423
+
async def store_then_patch(*_args, **_kwargs):
424
+
async with _db_session() as session:
425
+
await session.execute(
426
+
update(Track)
427
+
.where(Track.id == track_id)
428
+
.values(title="Patched Title")
429
+
)
430
+
await session.commit()
431
+
return storage_result(file_id="NEW")
432
+
433
+
with (
434
+
patched_replace_pipeline(
435
+
store=storage_result(file_id="NEW"),
436
+
update_record_side_effect=fake_update_record,
437
+
),
438
+
patch(
439
+
"backend.api.tracks.audio_replace._store_audio",
440
+
AsyncMock(side_effect=store_then_patch),
441
+
),
442
+
):
443
+
await _process_replace_background(replace_ctx(track_id=track_id))
444
+
445
+
# the published record reflects the PATCH'd title because we re-fetched
446
+
# state right before publishing
447
+
assert captured_record["title"] == "Patched Title"
448
+
449
+
450
+
def test_track_audio_state_dataclass() -> None:
451
+
"""sanity check the snapshot dataclass."""
452
+
state = TrackAudioState(
453
+
track_id=1,
454
+
artist_did=OWNER_DID,
455
+
artist_display_name="Owner",
456
+
atproto_record_uri=TRACK_URI,
457
+
old_file_id="OLD",
458
+
old_file_type="mp3",
459
+
old_original_file_id=None,
460
+
old_original_file_type=None,
461
+
title="My Song",
462
+
album=None,
463
+
duration=120,
464
+
features=[],
465
+
image_url=None,
466
+
description=None,
467
+
support_gate=None,
468
+
)
469
+
assert state.track_id == 1
470
+
assert state.support_gate is None
+14
backend/tests/conftest.py
+14
backend/tests/conftest.py
···
60
60
"""Mock delete."""
61
61
return True
62
62
63
+
async def delete_gated(self, file_id: str, file_type: str | None = None) -> bool:
64
+
"""Mock delete_gated."""
65
+
return True
66
+
67
+
async def save_gated(
68
+
self,
69
+
file: BinaryIO | BytesIO,
70
+
filename: str,
71
+
progress_callback: Callable[[float], None] | None = None,
72
+
) -> str:
73
+
"""Mock save_gated."""
74
+
return "mock_gated_file_id_456"
75
+
63
76
64
77
def pytest_configure(config):
65
78
"""Set mock storage before any test modules are imported."""
···
310
323
engine = create_async_engine(database_url, echo=False)
311
324
try:
312
325
async with engine.begin() as conn:
326
+
await conn.execute(sa.text("CREATE EXTENSION IF NOT EXISTS pg_trgm"))
313
327
await conn.run_sync(Base.metadata.create_all)
314
328
await _truncate_tables(conn)
315
329
await _create_clear_database_procedure(conn)
+8
-1
backend/tests/test_moderation.py
+8
-1
backend/tests/test_moderation.py
···
532
532
533
533
@pytest.fixture
534
534
def report_test_app() -> Generator[FastAPI, None, None]:
535
-
"""create test app with mocked auth for report tests."""
535
+
"""create test app with mocked auth for report tests.
536
+
537
+
also resets the slowapi rate-limiter so per-IP budgets from earlier tests
538
+
in the same session don't bleed into this one (causes spurious 429s under
539
+
parallel xdist runs).
540
+
"""
541
+
from backend.utilities.rate_limit import limiter
536
542
537
543
async def mock_require_auth() -> Session:
538
544
return MockReportSession()
539
545
540
546
app.dependency_overrides[require_auth] = mock_require_auth
547
+
limiter.reset()
541
548
542
549
yield app
543
550
+10
-2
loq.toml
+10
-2
loq.toml
···
48
48
49
49
[[rules]]
50
50
path = "backend/src/backend/storage/r2.py"
51
-
max_lines = 766
51
+
max_lines = 828
52
52
53
53
[[rules]]
54
54
path = "backend/tests/api/test_audio.py"
···
76
76
77
77
[[rules]]
78
78
path = "backend/tests/test_moderation.py"
79
-
max_lines = 715
79
+
max_lines = 718
80
80
81
81
[[rules]]
82
82
path = "docs/internal/authentication.md"
···
253
253
[[rules]]
254
254
path = "backend/src/backend/api/lists/playlists.py"
255
255
max_lines = 952
256
+
257
+
[[rules]]
258
+
path = "backend/src/backend/api/tracks/audio_replace.py"
259
+
max_lines = 651
260
+
261
+
[[rules]]
262
+
path = "backend/tests/conftest.py"
263
+
max_lines = 515