Skip to main content

max / makenotwork

8.1 KB · 273 lines History Blame Raw
1 //! Content insertion CRUD: reusable clips (intros, outros, sponsor reads)
2 //! and their per-item placements.
3
4 use sqlx::PgPool;
5
6 use super::models::{DbContentInsertion, DbInsertionPlacement, DbPlacementWithInsertion};
7 use super::{ContentInsertionId, ContentInsertionPlacementId, InsertionPosition, ItemId, UserId};
8 use crate::error::Result;
9
10 // ── Insertion library ──
11
12 /// Create a new reusable insertion clip for a creator.
13 #[allow(clippy::too_many_arguments)]
14 #[tracing::instrument(skip_all)]
15 pub(crate) async fn create_insertion(
16 pool: &PgPool,
17 user_id: UserId,
18 title: &str,
19 media_type: &str,
20 storage_key: &str,
21 duration_ms: i32,
22 file_size: i64,
23 mime_type: &str,
24 ) -> Result<DbContentInsertion> {
25 let row = sqlx::query_as::<_, DbContentInsertion>(
26 r"
27 INSERT INTO content_insertions (user_id, title, media_type, storage_key, duration_ms, file_size, mime_type)
28 VALUES ($1, $2, $3, $4, $5, $6, $7)
29 RETURNING *
30 ",
31 )
32 .bind(user_id)
33 .bind(title)
34 .bind(media_type)
35 .bind(storage_key)
36 .bind(duration_ms)
37 .bind(file_size)
38 .bind(mime_type)
39 .fetch_one(pool)
40 .await?;
41
42 Ok(row)
43 }
44
45 /// List all insertion clips for a creator, newest first.
46 #[tracing::instrument(skip_all)]
47 pub(crate) async fn list_insertions(
48 pool: &PgPool,
49 user_id: UserId,
50 ) -> Result<Vec<DbContentInsertion>> {
51 let rows = sqlx::query_as::<_, DbContentInsertion>(
52 "SELECT * FROM content_insertions WHERE user_id = $1 ORDER BY created_at DESC LIMIT 500",
53 )
54 .bind(user_id)
55 .fetch_all(pool)
56 .await?;
57
58 Ok(rows)
59 }
60
61 /// Get a single insertion by ID, scoped to the owning user.
62 #[tracing::instrument(skip_all)]
63 pub(crate) async fn get_insertion(
64 pool: &PgPool,
65 id: ContentInsertionId,
66 user_id: UserId,
67 ) -> Result<Option<DbContentInsertion>> {
68 let row = sqlx::query_as::<_, DbContentInsertion>(
69 "SELECT * FROM content_insertions WHERE id = $1 AND user_id = $2",
70 )
71 .bind(id)
72 .bind(user_id)
73 .fetch_optional(pool)
74 .await?;
75
76 Ok(row)
77 }
78
79 /// Get an insertion by its (user-scoped, deterministic) S3 storage key. Used to
80 /// make `confirm_insertion` idempotent: the key is `{user}/insertions/{filename}`
81 /// with no UUID and `storage_key` carries no UNIQUE, so a replayed confirm would
82 /// otherwise re-charge storage and insert a duplicate row (ultra-fuzz Run 12
83 /// Storage: confirm idempotency).
84 #[tracing::instrument(skip_all)]
85 pub(crate) async fn get_insertion_by_storage_key(
86 pool: &PgPool,
87 user_id: UserId,
88 storage_key: &str,
89 ) -> Result<Option<DbContentInsertion>> {
90 let row = sqlx::query_as::<_, DbContentInsertion>(
91 "SELECT * FROM content_insertions WHERE user_id = $1 AND storage_key = $2",
92 )
93 .bind(user_id)
94 .bind(storage_key)
95 .fetch_optional(pool)
96 .await?;
97
98 Ok(row)
99 }
100
101 /// Rename an insertion clip. Returns true if the row was found and updated.
102 #[tracing::instrument(skip_all)]
103 pub(crate) async fn update_insertion_title(
104 pool: &PgPool,
105 id: ContentInsertionId,
106 user_id: UserId,
107 title: &str,
108 ) -> Result<bool> {
109 let result =
110 sqlx::query("UPDATE content_insertions SET title = $3 WHERE id = $1 AND user_id = $2")
111 .bind(id)
112 .bind(user_id)
113 .bind(title)
114 .execute(pool)
115 .await?;
116
117 Ok(result.rows_affected() > 0)
118 }
119
120 /// Delete an insertion clip (placements cascade). Returns true if deleted.
121 #[tracing::instrument(skip_all)]
122 pub(crate) async fn delete_insertion(
123 pool: &PgPool,
124 id: ContentInsertionId,
125 user_id: UserId,
126 ) -> Result<bool> {
127 let result = sqlx::query("DELETE FROM content_insertions WHERE id = $1 AND user_id = $2")
128 .bind(id)
129 .bind(user_id)
130 .execute(pool)
131 .await?;
132
133 Ok(result.rows_affected() > 0)
134 }
135
136 // ── Placements ──
137
138 /// Attach an insertion clip to an item at a given position.
139 #[tracing::instrument(skip_all)]
140 pub(crate) async fn create_placement(
141 pool: &PgPool,
142 item_id: ItemId,
143 insertion_id: ContentInsertionId,
144 position: InsertionPosition,
145 offset_ms: Option<i32>,
146 sort_order: i32,
147 ) -> Result<DbInsertionPlacement> {
148 let row = sqlx::query_as::<_, DbInsertionPlacement>(
149 r"
150 INSERT INTO content_insertion_placements (item_id, insertion_id, position, offset_ms, sort_order)
151 VALUES ($1, $2, $3, $4, $5)
152 RETURNING *
153 ",
154 )
155 .bind(item_id)
156 .bind(insertion_id)
157 .bind(position)
158 .bind(offset_ms)
159 .bind(sort_order)
160 .fetch_one(pool)
161 .await?;
162
163 Ok(row)
164 }
165
166 /// List all placements for an item, joined with insertion metadata.
167 /// Ordered by position (pre_roll, mid_roll by offset, post_roll) then sort_order.
168 ///
169 /// This is the creator's placement-management view (it lists every placement the
170 /// creator configured, including ones whose clip is still pending/held), so it is
171 /// NOT scan-gated. The fan-facing playback path uses
172 /// [`list_playable_placements_for_item`], which hides un-cleared clips.
173 #[tracing::instrument(skip_all)]
174 pub(crate) async fn list_placements_for_item(
175 pool: &PgPool,
176 item_id: ItemId,
177 ) -> Result<Vec<DbPlacementWithInsertion>> {
178 let rows = sqlx::query_as::<_, DbPlacementWithInsertion>(
179 r"
180 SELECT
181 p.id, p.item_id, p.insertion_id, p.position, p.offset_ms, p.sort_order, p.created_at,
182 i.title AS insertion_title,
183 i.duration_ms AS insertion_duration_ms,
184 i.storage_key AS insertion_storage_key
185 FROM content_insertion_placements p
186 JOIN content_insertions i ON i.id = p.insertion_id
187 WHERE p.item_id = $1
188 ORDER BY
189 CASE p.position
190 WHEN 'pre_roll' THEN 0
191 WHEN 'mid_roll' THEN 1
192 WHEN 'post_roll' THEN 2
193 END,
194 p.offset_ms NULLS LAST,
195 p.sort_order
196 LIMIT 100
197 ",
198 )
199 .bind(item_id)
200 .fetch_all(pool)
201 .await?;
202
203 Ok(rows)
204 }
205
206 /// Fan-facing playback resolver: like [`list_placements_for_item`] but gated to
207 /// insertions whose scan cleared (`i.scan_status = 'clean'`). A pending or held
208 /// clip is never spliced into a fan's stream, the fail-closed gate for the
209 /// gate-less, CDN-served insertion kind. Used only by the public media player
210 /// (`build_segments_json`); creator management goes through the ungated variant.
211 #[tracing::instrument(skip_all)]
212 pub(crate) async fn list_playable_placements_for_item(
213 pool: &PgPool,
214 item_id: ItemId,
215 ) -> Result<Vec<DbPlacementWithInsertion>> {
216 let rows = sqlx::query_as::<_, DbPlacementWithInsertion>(
217 r"
218 SELECT
219 p.id, p.item_id, p.insertion_id, p.position, p.offset_ms, p.sort_order, p.created_at,
220 i.title AS insertion_title,
221 i.duration_ms AS insertion_duration_ms,
222 i.storage_key AS insertion_storage_key
223 FROM content_insertion_placements p
224 JOIN content_insertions i ON i.id = p.insertion_id
225 WHERE p.item_id = $1 AND i.scan_status = 'clean'
226 ORDER BY
227 CASE p.position
228 WHEN 'pre_roll' THEN 0
229 WHEN 'mid_roll' THEN 1
230 WHEN 'post_roll' THEN 2
231 END,
232 p.offset_ms NULLS LAST,
233 p.sort_order
234 LIMIT 100
235 ",
236 )
237 .bind(item_id)
238 .fetch_all(pool)
239 .await?;
240
241 Ok(rows)
242 }
243
244 /// Delete a single placement by ID. Returns true if deleted.
245 #[tracing::instrument(skip_all)]
246 pub(crate) async fn delete_placement(
247 pool: &PgPool,
248 placement_id: ContentInsertionPlacementId,
249 ) -> Result<bool> {
250 let result = sqlx::query("DELETE FROM content_insertion_placements WHERE id = $1")
251 .bind(placement_id)
252 .execute(pool)
253 .await?;
254
255 Ok(result.rows_affected() > 0)
256 }
257
258 /// Get a placement by ID (for ownership verification via item).
259 #[tracing::instrument(skip_all)]
260 pub(crate) async fn get_placement_by_id(
261 pool: &PgPool,
262 placement_id: ContentInsertionPlacementId,
263 ) -> Result<Option<DbInsertionPlacement>> {
264 let row = sqlx::query_as::<_, DbInsertionPlacement>(
265 "SELECT * FROM content_insertion_placements WHERE id = $1",
266 )
267 .bind(placement_id)
268 .fetch_optional(pool)
269 .await?;
270
271 Ok(row)
272 }
273