Skip to main content

max / makenotwork

19.8 KB · 613 lines History Blame Raw
1 //! Content insertion API: reusable clip library + per-item placement management.
2
3 use axum::{
4 Json,
5 extract::{Path, State},
6 response::{Html, IntoResponse},
7 };
8 use serde::{Deserialize, Serialize};
9
10 use crate::{AppStorage, Scanning};
11 use sqlx::PgPool;
12
13 use crate::{
14 auth::AuthUser,
15 db::{self, ContentInsertionId, ContentInsertionPlacementId, InsertionPosition, ItemId},
16 error::{AppError, Result, ResultExt},
17 helpers::htmx_toast_response,
18 routes::storage::{CommitTarget, commit_upload},
19 storage::{FileType, S3Client},
20 templates::InsertionListTemplate,
21 };
22
23 use super::verify_item_ownership;
24
25 // Request/Response Types
26
27 #[derive(Debug, Deserialize)]
28 pub(super) struct InsertionPresignRequest {
29 pub file_name: String,
30 pub content_type: String,
31 }
32
33 #[derive(Debug, Serialize)]
34 pub(super) struct InsertionPresignResponse {
35 pub upload_url: String,
36 pub s3_key: String,
37 pub expires_in: u64,
38 }
39
40 #[derive(Debug, Deserialize)]
41 pub(super) struct InsertionConfirmRequest {
42 pub s3_key: String,
43 pub title: String,
44 pub duration_ms: i32,
45 #[allow(dead_code)] // kept for client compat; real size fetched from S3
46 pub file_size: i64,
47 pub mime_type: String,
48 }
49
50 #[derive(Debug, Serialize)]
51 pub(super) struct InsertionResponse {
52 pub id: ContentInsertionId,
53 pub title: String,
54 pub media_type: String,
55 pub duration_ms: i32,
56 pub file_size: i64,
57 }
58
59 #[derive(Debug, Deserialize)]
60 pub(super) struct RenameInsertionRequest {
61 pub title: String,
62 }
63
64 #[derive(Debug, Deserialize)]
65 pub(super) struct CreatePlacementRequest {
66 pub insertion_id: ContentInsertionId,
67 pub position: InsertionPosition,
68 pub offset_ms: Option<i32>,
69 #[serde(default)]
70 pub sort_order: i32,
71 }
72
73 // Insertion Library Handlers
74
75 /// Generate a presigned URL for uploading an insertion clip.
76 ///
77 /// POST /api/users/me/insertions/presign
78 #[tracing::instrument(skip_all, name = "insertions::presign")]
79 pub(super) async fn presign_insertion(
80 State(db): State<PgPool>,
81 State(storage): State<AppStorage>,
82 AuthUser(user): AuthUser,
83 Json(req): Json<InsertionPresignRequest>,
84 ) -> Result<impl IntoResponse> {
85 user.check_not_suspended()?;
86 let s3 = storage.require_s3()?;
87
88 S3Client::validate_content_type(FileType::Insertion, &req.content_type)?;
89 S3Client::validate_extension(FileType::Insertion, &req.file_name)?;
90
91 // Check storage quota before issuing presigned URL
92 db::creator_tiers::check_presign_allowed(&db, user.id, FileType::Insertion).await?;
93
94 // Staging key (unserved); the scan worker promotes it to the content key on a
95 // Clean verdict (C1). Insertion clips are served presigned from the stored
96 // key, so promotion only repoints `content_insertions.storage_key`.
97 let s3_key = S3Client::generate_staging_key(&req.file_name);
98
99 // Track the pending upload so the reaper can clean it up if never confirmed
100 db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
101
102 let expires_in = 3600;
103 let upload_url = s3
104 .presign_upload(
105 &s3_key,
106 &req.content_type,
107 Some(expires_in),
108 Some(crate::storage::CACHE_CONTROL_IMMUTABLE),
109 None,
110 )
111 .await
112 .context("presign upload for insertion clip")?;
113
114 Ok(Json(InsertionPresignResponse {
115 upload_url,
116 s3_key: s3_key.into_string(),
117 expires_in,
118 }))
119 }
120
121 /// Confirm an insertion upload and create the DB record.
122 ///
123 /// POST /api/users/me/insertions/confirm
124 #[tracing::instrument(skip_all, name = "insertions::confirm")]
125 pub(super) async fn confirm_insertion(
126 State(db): State<PgPool>,
127 State(storage): State<AppStorage>,
128 State(scanning): State<Scanning>,
129 AuthUser(user): AuthUser,
130 Json(req): Json<InsertionConfirmRequest>,
131 ) -> Result<impl IntoResponse> {
132 user.check_not_suspended()?;
133 let s3 = storage.require_s3()?;
134
135 // Ownership of the staging key is proved below (after the replay
136 // short-circuit) via `pending_uploads`, a `staging/{uuid}` key carries no
137 // user in its path for a prefix check to bind against.
138
139 // Idempotent replay: a retried confirm for the same key must not
140 // re-charge storage or insert a duplicate row (`storage_key` has no UNIQUE).
141 // Return the already-created insertion (ultra-fuzz Run 12 Storage: confirm
142 // idempotency).
143 if let Some(existing) =
144 db::content_insertions::get_insertion_by_storage_key(&db, user.id, &req.s3_key).await?
145 {
146 return Ok(Json(InsertionResponse {
147 id: existing.id,
148 title: existing.title,
149 media_type: existing.media_type,
150 duration_ms: existing.duration_ms,
151 file_size: existing.file_size,
152 }));
153 }
154
155 // Authorize the staging key for a fresh confirm: the caller must have
156 // presigned it (recorded against them in `pending_uploads`). Placed after the
157 // replay short-circuit, which is authorized by the row already existing,
158 // and before the size/tier reject paths, so an unowned (at most another
159 // user's in-flight) staging object is never enqueued for deletion.
160 if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
161 return Err(AppError::BadRequest("Invalid upload key".to_string()));
162 }
163
164 // Get real file size from S3 (never trust client-provided size)
165 let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
166 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
167 })?;
168
169 // Enforce static per-type size limit
170 if file_size_bytes as u64 > FileType::Insertion.max_size() {
171 crate::routes::storage::enqueue_s3_orphan(
172 &db,
173 &req.s3_key,
174 crate::storage::S3Bucket::Main,
175 "insertion_upload_rejected",
176 )
177 .await;
178 return Err(AppError::BadRequest(format!(
179 "File exceeds maximum size of {} MB",
180 FileType::Insertion.max_size() / (1024 * 1024)
181 )));
182 }
183
184 // Validate mime_type at confirm time (client could change it after presign)
185 S3Client::validate_content_type(FileType::Insertion, &req.mime_type)?;
186
187 if req.title.is_empty() || req.title.len() > 200 {
188 return Err(AppError::BadRequest(
189 "Title must be 1-200 characters".to_string(),
190 ));
191 }
192 if req.duration_ms <= 0 {
193 return Err(AppError::BadRequest(
194 "Duration must be positive".to_string(),
195 ));
196 }
197
198 // Enforce tier-based limits (per-file + storage cap)
199 let max_storage = match db::creator_tiers::check_upload_allowed(
200 &db,
201 user.id,
202 FileType::Insertion,
203 file_size_bytes,
204 )
205 .await
206 {
207 Ok(max) => max,
208 Err(e) => {
209 crate::routes::storage::enqueue_s3_orphan(
210 &db,
211 &req.s3_key,
212 crate::storage::S3Bucket::Main,
213 "insertion_upload_rejected",
214 )
215 .await;
216 return Err(e);
217 }
218 };
219
220 // Atomically increment storage BEFORE writing the DB record.
221 // Avoids orphaned unbilled file references.
222 if let Err(e) =
223 db::creator_tiers::try_increment_storage(&db, user.id, file_size_bytes, max_storage).await
224 {
225 crate::routes::storage::enqueue_s3_orphan(
226 &db,
227 &req.s3_key,
228 crate::storage::S3Bucket::Main,
229 "insertion_upload_rejected",
230 )
231 .await;
232 return Err(e);
233 }
234
235 // Clear the pending upload record now that the upload is confirmed
236 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
237
238 // Persist the real media family (audio vs video) derived from the validated
239 // MIME, rather than assuming audio. Drives the placement type-compat guard
240 // and which player element the clip renders in.
241 let media_type = S3Client::insertion_media_type(&req.mime_type);
242
243 let insertion = db::content_insertions::create_insertion(
244 &db,
245 user.id,
246 &req.title,
247 media_type,
248 &req.s3_key,
249 req.duration_ms,
250 file_size_bytes,
251 &req.mime_type,
252 )
253 .await?;
254
255 // Scan AFTER the insertion row is created, same ordering rule that the
256 // storage handlers follow. The row starts scan_status='pending' (fail-closed
257 // gate): fan playback (list_playable_placements_for_item) hides it until the
258 // worker flips it to 'clean'; on quarantine the row is purged and a WAM ticket
259 // filed. The creator still sees the pending clip in their management library.
260 commit_upload(
261 &db,
262 scanning.scanner.as_ref(),
263 CommitTarget::ContentInsertion(insertion.id),
264 &req.s3_key,
265 FileType::Insertion,
266 user.id,
267 file_size_bytes,
268 )
269 .await?;
270
271 tracing::info!(
272 "Insertion confirmed: id={}, user={}, key={}",
273 insertion.id,
274 user.id,
275 req.s3_key
276 );
277
278 Ok(Json(InsertionResponse {
279 id: insertion.id,
280 title: insertion.title,
281 media_type: insertion.media_type,
282 duration_ms: insertion.duration_ms,
283 file_size: insertion.file_size,
284 }))
285 }
286
287 /// List all insertion clips for the current user (HTMX partial).
288 ///
289 /// GET /api/users/me/insertions
290 #[tracing::instrument(skip_all, name = "insertions::list")]
291 pub(super) async fn list_insertions(
292 State(db): State<PgPool>,
293 AuthUser(user): AuthUser,
294 ) -> Result<impl IntoResponse> {
295 let insertions = db::content_insertions::list_insertions(&db, user.id).await?;
296
297 let display: Vec<crate::templates::InsertionDisplay> = insertions
298 .iter()
299 .map(|i| crate::templates::InsertionDisplay {
300 id: i.id.to_string(),
301 title: i.title.clone(),
302 media_type: i.media_type.clone(),
303 duration_display: format_duration_ms(i.duration_ms),
304 created_at: i.created_at.format("%Y-%m-%d").to_string(),
305 })
306 .collect();
307
308 Ok(Html(crate::helpers::render_fragment(
309 &InsertionListTemplate {
310 insertions: display,
311 },
312 )?))
313 }
314
315 /// Rename an insertion clip.
316 ///
317 /// PUT /api/insertions/{id}
318 #[tracing::instrument(skip_all, name = "insertions::rename")]
319 pub(super) async fn rename_insertion(
320 State(db): State<PgPool>,
321 AuthUser(user): AuthUser,
322 Path(id): Path<ContentInsertionId>,
323 Json(req): Json<RenameInsertionRequest>,
324 ) -> Result<impl IntoResponse> {
325 user.check_not_suspended()?;
326 if req.title.is_empty() || req.title.len() > 200 {
327 return Err(AppError::BadRequest(
328 "Title must be 1-200 characters".to_string(),
329 ));
330 }
331
332 let updated =
333 db::content_insertions::update_insertion_title(&db, id, user.id, &req.title).await?;
334
335 if !updated {
336 return Err(AppError::NotFound);
337 }
338
339 Ok(htmx_toast_response("Clip renamed", "success"))
340 }
341
342 /// Delete an insertion clip (cascades placements).
343 ///
344 /// DELETE /api/insertions/{id}
345 #[tracing::instrument(skip_all, name = "insertions::delete")]
346 pub(super) async fn delete_insertion(
347 State(db): State<PgPool>,
348 AuthUser(user): AuthUser,
349 Path(id): Path<ContentInsertionId>,
350 ) -> Result<impl IntoResponse> {
351 user.check_not_suspended()?;
352 // Look up the insertion for S3 cleanup and storage decrement
353 let insertion = db::content_insertions::get_insertion(&db, id, user.id).await?;
354 let file_size = insertion.as_ref().map_or(0, |i| i.file_size);
355
356 // Enqueue as the sole durable deletion path, BEFORE the row delete. Abort on
357 // enqueue failure rather than warn-and-proceed: deleting the row anyway would
358 // orphan the clip's S3 object with no record (ultra-fuzz Run 12 Storage F3).
359 // The reaper's is_s3_key_live guard makes the reverse case safe.
360 if let Some(ref ins) = insertion {
361 db::pending_s3_deletions::enqueue_deletions(
362 &db,
363 &[(ins.storage_key.clone(), "main".to_string())],
364 "insertion_delete",
365 )
366 .await?;
367 }
368
369 let deleted = db::content_insertions::delete_insertion(&db, id, user.id).await?;
370
371 if !deleted {
372 return Err(AppError::NotFound);
373 }
374
375 // Decrement storage counter
376 if file_size > 0 {
377 db::creator_tiers::decrement_storage_used(&db, user.id, file_size).await?;
378 }
379
380 Ok(htmx_toast_response("Clip deleted", "success"))
381 }
382
383 // Placement Handlers
384
385 /// List placements for an item (HTMX partial).
386 ///
387 /// GET /api/items/{id}/insertions
388 #[tracing::instrument(skip_all, name = "insertions::list_placements")]
389 pub(super) async fn list_placements(
390 State(db): State<PgPool>,
391 AuthUser(user): AuthUser,
392 Path(item_id): Path<ItemId>,
393 ) -> Result<impl IntoResponse> {
394 let (item, _project) = verify_item_ownership(&db, item_id, user.id).await?;
395
396 let placements = db::content_insertions::list_placements_for_item(&db, item_id).await?;
397 // Only offer clips that can legally be placed on this item (a video clip on an
398 // audio item would be rejected by create_placement, so don't surface it).
399 let available: Vec<_> = db::content_insertions::list_insertions(&db, user.id)
400 .await?
401 .into_iter()
402 .filter(|i| clip_compatible_with_item(&i.media_type, item.item_type))
403 .collect();
404
405 let placement_display: Vec<crate::templates::PlacementDisplay> = placements
406 .iter()
407 .map(|p| crate::templates::PlacementDisplay {
408 id: p.id.to_string(),
409 insertion_title: p.insertion_title.clone(),
410 position: p.position.to_string(),
411 offset_display: p.offset_ms.map(format_duration_ms),
412 sort_order: p.sort_order,
413 })
414 .collect();
415
416 let insertion_display: Vec<crate::templates::InsertionDisplay> = available
417 .iter()
418 .map(|i| crate::templates::InsertionDisplay {
419 id: i.id.to_string(),
420 title: i.title.clone(),
421 media_type: i.media_type.clone(),
422 duration_display: format_duration_ms(i.duration_ms),
423 created_at: i.created_at.format("%Y-%m-%d").to_string(),
424 })
425 .collect();
426
427 Ok(Html(crate::helpers::render_fragment(
428 &crate::templates::PlacementListTemplate {
429 item_id: item_id.to_string(),
430 placements: placement_display,
431 available_insertions: insertion_display,
432 },
433 )?))
434 }
435
436 /// Create a placement (attach an insertion to an item).
437 ///
438 /// POST /api/items/{id}/insertions
439 #[tracing::instrument(skip_all, name = "insertions::create_placement")]
440 pub(super) async fn create_placement(
441 State(db): State<PgPool>,
442 AuthUser(user): AuthUser,
443 Path(item_id): Path<ItemId>,
444 Json(req): Json<CreatePlacementRequest>,
445 ) -> Result<impl IntoResponse> {
446 user.check_not_suspended()?;
447 let (item, _project) = verify_item_ownership(&db, item_id, user.id).await?;
448
449 // Verify the insertion belongs to this user
450 let insertion = db::content_insertions::get_insertion(&db, req.insertion_id, user.id)
451 .await?
452 .ok_or(AppError::NotFound)?;
453
454 // Only audio/video items host a media player, so only they can carry clips.
455 // A video clip additionally requires a video item, an audio item renders in
456 // an `<audio>` element that would silently drop the clip's video track. An
457 // audio clip on a video item is allowed (plays as a blank-frame segment).
458 if !item_hosts_insertions(item.item_type) {
459 return Err(AppError::BadRequest(
460 "Clips can only be placed on audio or video items".to_string(),
461 ));
462 }
463 if !clip_compatible_with_item(&insertion.media_type, item.item_type) {
464 return Err(AppError::BadRequest(
465 "Video clips can only be placed on video items".to_string(),
466 ));
467 }
468
469 // Validate mid-roll offset
470 if req.position == InsertionPosition::MidRoll && req.offset_ms.is_none() {
471 return Err(AppError::BadRequest(
472 "Mid-roll clips require an offset".to_string(),
473 ));
474 }
475
476 let _placement = db::content_insertions::create_placement(
477 &db,
478 item_id,
479 insertion.id,
480 req.position,
481 req.offset_ms,
482 req.sort_order,
483 )
484 .await?;
485
486 Ok(htmx_toast_response("Clip added", "success"))
487 }
488
489 /// Remove a placement.
490 ///
491 /// DELETE /api/item-insertions/{id}
492 #[tracing::instrument(skip_all, name = "insertions::delete_placement")]
493 pub(super) async fn delete_placement(
494 State(db): State<PgPool>,
495 AuthUser(user): AuthUser,
496 Path(placement_id): Path<ContentInsertionPlacementId>,
497 ) -> Result<impl IntoResponse> {
498 user.check_not_suspended()?;
499 // Get placement to verify item ownership
500 let placement = db::content_insertions::get_placement_by_id(&db, placement_id)
501 .await?
502 .ok_or(AppError::NotFound)?;
503
504 verify_item_ownership(&db, placement.item_id, user.id).await?;
505
506 db::content_insertions::delete_placement(&db, placement_id).await?;
507
508 Ok(htmx_toast_response("Clip removed", "success"))
509 }
510
511 // Helpers
512
513 /// Whether an item type has a media player that can host insertion clips.
514 /// Only audio and video items render a segment-aware player.
515 fn item_hosts_insertions(item_type: db::ItemType) -> bool {
516 matches!(item_type, db::ItemType::Audio | db::ItemType::Video)
517 }
518
519 /// Whether a clip of the given media type may be placed on the given item type.
520 /// Video clips require a video item; audio clips fit either audio or video items.
521 fn clip_compatible_with_item(clip_media_type: &str, item_type: db::ItemType) -> bool {
522 if !item_hosts_insertions(item_type) {
523 return false;
524 }
525 clip_media_type != "video" || item_type == db::ItemType::Video
526 }
527
528 /// Format milliseconds as MM:SS.
529 fn format_duration_ms(ms: i32) -> String {
530 // Clamp negatives to zero: duration_ms is validated > 0 at confirm time, so
531 // this is unreachable in practice, but a stray negative previously produced
532 // a malformed "0:-1" rather than a sane "0:00".
533 let total_secs = ms.max(0) / 1000;
534 let mins = total_secs / 60;
535 let secs = total_secs % 60;
536 format!("{mins}:{secs:02}")
537 }
538
539 #[cfg(test)]
540 mod tests {
541 use super::*;
542
543 #[test]
544 fn zero_ms() {
545 assert_eq!(format_duration_ms(0), "0:00");
546 }
547
548 #[test]
549 fn one_second() {
550 assert_eq!(format_duration_ms(1000), "0:01");
551 }
552
553 #[test]
554 fn fifty_nine_seconds() {
555 assert_eq!(format_duration_ms(59000), "0:59");
556 }
557
558 #[test]
559 fn one_minute() {
560 assert_eq!(format_duration_ms(60000), "1:00");
561 }
562
563 #[test]
564 fn one_minute_one_second() {
565 assert_eq!(format_duration_ms(61000), "1:01");
566 }
567
568 #[test]
569 fn over_sixty_minutes() {
570 assert_eq!(format_duration_ms(3_661_000), "61:01");
571 }
572
573 #[test]
574 fn sub_second_rounds_down() {
575 assert_eq!(format_duration_ms(500), "0:00");
576 }
577
578 #[test]
579 fn negative_ms_clamps_to_zero() {
580 // Defensive clamp: a negative duration (unreachable, validated > 0 at
581 // confirm) formats as "0:00", not a malformed "0:-1".
582 assert_eq!(format_duration_ms(-1500), "0:00");
583 assert_eq!(format_duration_ms(-1), "0:00");
584 }
585
586 #[test]
587 fn only_audio_and_video_items_host_insertions() {
588 assert!(item_hosts_insertions(db::ItemType::Audio));
589 assert!(item_hosts_insertions(db::ItemType::Video));
590 assert!(!item_hosts_insertions(db::ItemType::Text));
591 assert!(!item_hosts_insertions(db::ItemType::Digital));
592 assert!(!item_hosts_insertions(db::ItemType::Bundle));
593 }
594
595 #[test]
596 fn audio_clip_fits_audio_and_video_items() {
597 assert!(clip_compatible_with_item("audio", db::ItemType::Audio));
598 assert!(clip_compatible_with_item("audio", db::ItemType::Video));
599 }
600
601 #[test]
602 fn video_clip_requires_video_item() {
603 assert!(clip_compatible_with_item("video", db::ItemType::Video));
604 assert!(!clip_compatible_with_item("video", db::ItemType::Audio));
605 }
606
607 #[test]
608 fn no_clip_fits_a_non_media_item() {
609 assert!(!clip_compatible_with_item("audio", db::ItemType::Text));
610 assert!(!clip_compatible_with_item("video", db::ItemType::Text));
611 }
612 }
613