Skip to main content

max / makenotwork

8.1 KB · 272 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.
83 #[tracing::instrument(skip_all)]
84 pub(crate) async fn get_insertion_by_storage_key(
85 pool: &PgPool,
86 user_id: UserId,
87 storage_key: &str,
88 ) -> Result<Option<DbContentInsertion>> {
89 let row = sqlx::query_as::<_, DbContentInsertion>(
90 "SELECT * FROM content_insertions WHERE user_id = $1 AND storage_key = $2",
91 )
92 .bind(user_id)
93 .bind(storage_key)
94 .fetch_optional(pool)
95 .await?;
96
97 Ok(row)
98 }
99
100 /// Rename an insertion clip. Returns true if the row was found and updated.
101 #[tracing::instrument(skip_all)]
102 pub(crate) async fn update_insertion_title(
103 pool: &PgPool,
104 id: ContentInsertionId,
105 user_id: UserId,
106 title: &str,
107 ) -> Result<bool> {
108 let result =
109 sqlx::query("UPDATE content_insertions SET title = $3 WHERE id = $1 AND user_id = $2")
110 .bind(id)
111 .bind(user_id)
112 .bind(title)
113 .execute(pool)
114 .await?;
115
116 Ok(result.rows_affected() > 0)
117 }
118
119 /// Delete an insertion clip (placements cascade). Returns true if deleted.
120 #[tracing::instrument(skip_all)]
121 pub(crate) async fn delete_insertion(
122 pool: &PgPool,
123 id: ContentInsertionId,
124 user_id: UserId,
125 ) -> Result<bool> {
126 let result = sqlx::query("DELETE FROM content_insertions WHERE id = $1 AND user_id = $2")
127 .bind(id)
128 .bind(user_id)
129 .execute(pool)
130 .await?;
131
132 Ok(result.rows_affected() > 0)
133 }
134
135 // ── Placements ──
136
137 /// Attach an insertion clip to an item at a given position.
138 #[tracing::instrument(skip_all)]
139 pub(crate) async fn create_placement(
140 pool: &PgPool,
141 item_id: ItemId,
142 insertion_id: ContentInsertionId,
143 position: InsertionPosition,
144 offset_ms: Option<i32>,
145 sort_order: i32,
146 ) -> Result<DbInsertionPlacement> {
147 let row = sqlx::query_as::<_, DbInsertionPlacement>(
148 r"
149 INSERT INTO content_insertion_placements (item_id, insertion_id, position, offset_ms, sort_order)
150 VALUES ($1, $2, $3, $4, $5)
151 RETURNING *
152 ",
153 )
154 .bind(item_id)
155 .bind(insertion_id)
156 .bind(position)
157 .bind(offset_ms)
158 .bind(sort_order)
159 .fetch_one(pool)
160 .await?;
161
162 Ok(row)
163 }
164
165 /// List all placements for an item, joined with insertion metadata.
166 /// Ordered by position (pre_roll, mid_roll by offset, post_roll) then sort_order.
167 ///
168 /// This is the creator's placement-management view (it lists every placement the
169 /// creator configured, including ones whose clip is still pending/held), so it is
170 /// NOT scan-gated. The fan-facing playback path uses
171 /// [`list_playable_placements_for_item`], which hides un-cleared clips.
172 #[tracing::instrument(skip_all)]
173 pub(crate) async fn list_placements_for_item(
174 pool: &PgPool,
175 item_id: ItemId,
176 ) -> Result<Vec<DbPlacementWithInsertion>> {
177 let rows = sqlx::query_as::<_, DbPlacementWithInsertion>(
178 r"
179 SELECT
180 p.id, p.item_id, p.insertion_id, p.position, p.offset_ms, p.sort_order, p.created_at,
181 i.title AS insertion_title,
182 i.duration_ms AS insertion_duration_ms,
183 i.storage_key AS insertion_storage_key
184 FROM content_insertion_placements p
185 JOIN content_insertions i ON i.id = p.insertion_id
186 WHERE p.item_id = $1
187 ORDER BY
188 CASE p.position
189 WHEN 'pre_roll' THEN 0
190 WHEN 'mid_roll' THEN 1
191 WHEN 'post_roll' THEN 2
192 END,
193 p.offset_ms NULLS LAST,
194 p.sort_order
195 LIMIT 100
196 ",
197 )
198 .bind(item_id)
199 .fetch_all(pool)
200 .await?;
201
202 Ok(rows)
203 }
204
205 /// Fan-facing playback resolver: like [`list_placements_for_item`] but gated to
206 /// insertions whose scan cleared (`i.scan_status = 'clean'`). A pending or held
207 /// clip is never spliced into a fan's stream, the fail-closed gate for the
208 /// gate-less, CDN-served insertion kind. Used only by the public media player
209 /// (`build_segments_json`); creator management goes through the ungated variant.
210 #[tracing::instrument(skip_all)]
211 pub(crate) async fn list_playable_placements_for_item(
212 pool: &PgPool,
213 item_id: ItemId,
214 ) -> Result<Vec<DbPlacementWithInsertion>> {
215 let rows = sqlx::query_as::<_, DbPlacementWithInsertion>(
216 r"
217 SELECT
218 p.id, p.item_id, p.insertion_id, p.position, p.offset_ms, p.sort_order, p.created_at,
219 i.title AS insertion_title,
220 i.duration_ms AS insertion_duration_ms,
221 i.storage_key AS insertion_storage_key
222 FROM content_insertion_placements p
223 JOIN content_insertions i ON i.id = p.insertion_id
224 WHERE p.item_id = $1 AND i.scan_status = 'clean'
225 ORDER BY
226 CASE p.position
227 WHEN 'pre_roll' THEN 0
228 WHEN 'mid_roll' THEN 1
229 WHEN 'post_roll' THEN 2
230 END,
231 p.offset_ms NULLS LAST,
232 p.sort_order
233 LIMIT 100
234 ",
235 )
236 .bind(item_id)
237 .fetch_all(pool)
238 .await?;
239
240 Ok(rows)
241 }
242
243 /// Delete a single placement by ID. Returns true if deleted.
244 #[tracing::instrument(skip_all)]
245 pub(crate) async fn delete_placement(
246 pool: &PgPool,
247 placement_id: ContentInsertionPlacementId,
248 ) -> Result<bool> {
249 let result = sqlx::query("DELETE FROM content_insertion_placements WHERE id = $1")
250 .bind(placement_id)
251 .execute(pool)
252 .await?;
253
254 Ok(result.rows_affected() > 0)
255 }
256
257 /// Get a placement by ID (for ownership verification via item).
258 #[tracing::instrument(skip_all)]
259 pub(crate) async fn get_placement_by_id(
260 pool: &PgPool,
261 placement_id: ContentInsertionPlacementId,
262 ) -> Result<Option<DbInsertionPlacement>> {
263 let row = sqlx::query_as::<_, DbInsertionPlacement>(
264 "SELECT * FROM content_insertion_placements WHERE id = $1",
265 )
266 .bind(placement_id)
267 .fetch_optional(pool)
268 .await?;
269
270 Ok(row)
271 }
272