Skip to main content

max / makenotwork

13.2 KB · 394 lines History Blame Raw
1 //! Project member and revenue split operations.
2
3 use sqlx::PgPool;
4
5 use super::ProjectRole;
6 use super::id_types::{ProjectId, TipId, TransactionId, UserId};
7 use super::models::{DbProjectMember, DbProjectMemberWithUser, DbRevenueSplit, DbSplitExportRow};
8 use crate::error::{AppError, Result};
9
10 // ── Project Members ──
11
12 /// Take a row lock on the parent `projects` row so every revenue-split mutation
13 /// for the project serializes, including the first two members added
14 /// concurrently, which member-row locks alone would not block (audit Run 13
15 /// Conc TOCTOU). Callers must already be inside a transaction.
16 async fn lock_project_for_splits(
17 tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
18 project_id: ProjectId,
19 ) -> Result<()> {
20 sqlx::query("SELECT 1 FROM projects WHERE id = $1 FOR UPDATE")
21 .bind(project_id)
22 .execute(&mut **tx)
23 .await?;
24 Ok(())
25 }
26
27 /// Add a member to a project with a revenue split percentage (0-100).
28 ///
29 /// The sum of all member splits (excluding the owner's implicit remainder)
30 /// must not exceed 100.
31 #[tracing::instrument(skip(pool))]
32 pub(crate) async fn add_project_member(
33 pool: &PgPool,
34 project_id: ProjectId,
35 user_id: UserId,
36 role: ProjectRole,
37 split_percent: i16,
38 added_by: UserId,
39 ) -> Result<DbProjectMember> {
40 // Reject negative and out-of-range splits before any DB write. The Run #6
41 // audit caught the missing lower bound, negative percentages flow through
42 // `compute_splits` and record negative obligations.
43 if !(0..=100).contains(&split_percent) {
44 return Err(AppError::BadRequest(format!(
45 "split_percent must be between 0 and 100 (got {split_percent})"
46 )));
47 }
48
49 let mut tx = pool.begin().await?;
50
51 // Serialize every split mutation for this project on the parent row. Locking
52 // only the member rows does NOT block a concurrent INSERT of a new member
53 // when none exist yet, so two "first members" could each pass the cap check
54 // and race the total past 100% (audit Run 13 Conc TOCTOU). The projects row
55 // always exists, so `FOR UPDATE` on it serializes all adds/updates.
56 lock_project_for_splits(&mut tx, project_id).await?;
57
58 // Subtract the existing row's split (if this is an upsert) before the cap
59 // check; otherwise a legitimate update is rejected as "> 100%" whenever
60 // current_split + new_split crosses 100 even if the post-update total wouldn't.
61 let existing_split: (Option<i16>,) = sqlx::query_as(
62 "SELECT split_percent FROM project_members WHERE project_id = $1 AND user_id = $2",
63 )
64 .bind(project_id)
65 .bind(user_id)
66 .fetch_optional(&mut *tx)
67 .await?
68 .map_or((None,), |r: (i16,)| (Some(r.0),));
69 let existing = existing_split.0.unwrap_or(0) as i64;
70
71 let current_total: (Option<i64>,) = sqlx::query_as(
72 "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
73 )
74 .bind(project_id)
75 .fetch_one(&mut *tx)
76 .await?;
77
78 let total = current_total.0.unwrap_or(0);
79 let new_total = total - existing + split_percent as i64;
80 if new_total > 100 {
81 return Err(AppError::BadRequest(format!(
82 "Total split would be {new_total}%, exceeding 100%"
83 )));
84 }
85
86 let member = sqlx::query_as::<_, DbProjectMember>(
87 r"
88 INSERT INTO project_members (project_id, user_id, role, split_percent, added_by)
89 VALUES ($1, $2, $3, $4, $5)
90 ON CONFLICT (project_id, user_id) DO UPDATE
91 SET role = EXCLUDED.role,
92 split_percent = EXCLUDED.split_percent
93 RETURNING *
94 ",
95 )
96 .bind(project_id)
97 .bind(user_id)
98 .bind(role)
99 .bind(split_percent)
100 .bind(added_by)
101 .fetch_one(&mut *tx)
102 .await?;
103
104 tx.commit().await?;
105 Ok(member)
106 }
107
108 /// Remove a member from a project.
109 #[tracing::instrument(skip(pool))]
110 pub(crate) async fn remove_project_member(
111 pool: &PgPool,
112 project_id: ProjectId,
113 user_id: UserId,
114 ) -> Result<bool> {
115 let result = sqlx::query("DELETE FROM project_members WHERE project_id = $1 AND user_id = $2")
116 .bind(project_id)
117 .bind(user_id)
118 .execute(pool)
119 .await?;
120
121 Ok(result.rows_affected() > 0)
122 }
123
124 /// Get all members of a project with user info, ordered by split descending.
125 #[tracing::instrument(skip(pool))]
126 pub(crate) async fn get_project_members(
127 pool: &PgPool,
128 project_id: ProjectId,
129 ) -> Result<Vec<DbProjectMemberWithUser>> {
130 let members = sqlx::query_as::<_, DbProjectMemberWithUser>(
131 r"
132 SELECT pm.id, pm.project_id, pm.user_id, pm.role, pm.split_percent, pm.added_at,
133 u.username, u.display_name, u.stripe_account_id, u.stripe_charges_enabled
134 FROM project_members pm
135 JOIN users u ON u.id = pm.user_id
136 WHERE pm.project_id = $1
137 ORDER BY pm.split_percent DESC
138 ",
139 )
140 .bind(project_id)
141 .fetch_all(pool)
142 .await?;
143
144 Ok(members)
145 }
146
147 /// Get the total split percentage allocated to members (excluding the owner).
148 #[tracing::instrument(skip(pool))]
149 pub(crate) async fn get_total_split_percent(pool: &PgPool, project_id: ProjectId) -> Result<i64> {
150 let row: (Option<i64>,) = sqlx::query_as(
151 "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
152 )
153 .bind(project_id)
154 .fetch_one(pool)
155 .await?;
156
157 Ok(row.0.unwrap_or(0))
158 }
159
160 /// Update a member's split percentage.
161 #[allow(dead_code)]
162 #[tracing::instrument(skip(pool))]
163 pub(crate) async fn update_member_split(
164 pool: &PgPool,
165 project_id: ProjectId,
166 user_id: UserId,
167 new_split_percent: i16,
168 ) -> Result<()> {
169 if !(0..=100).contains(&new_split_percent) {
170 return Err(AppError::BadRequest(format!(
171 "split_percent must be between 0 and 100 (got {new_split_percent})"
172 )));
173 }
174 let mut tx = pool.begin().await?;
175
176 // Serialize on the parent project row (see `add_project_member`). The prior
177 // `SUM(...) FOR UPDATE` below was invalid, Postgres rejects FOR UPDATE with
178 // an aggregate, so this function 500'd on every call (audit Run 13); the lock
179 // now lives on the projects row and the SUM runs plain.
180 lock_project_for_splits(&mut tx, project_id).await?;
181
182 let current: Option<DbProjectMember> =
183 sqlx::query_as("SELECT * FROM project_members WHERE project_id = $1 AND user_id = $2")
184 .bind(project_id)
185 .bind(user_id)
186 .fetch_optional(&mut *tx)
187 .await?;
188
189 let current_member_split = current.map_or(0, |m| m.split_percent as i64);
190
191 let total_row: (Option<i64>,) = sqlx::query_as(
192 "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
193 )
194 .bind(project_id)
195 .fetch_one(&mut *tx)
196 .await?;
197
198 let total = total_row.0.unwrap_or(0);
199 let new_total = total - current_member_split + new_split_percent as i64;
200 if new_total > 100 {
201 return Err(AppError::BadRequest(format!(
202 "Total split would be {new_total}%, exceeding 100%"
203 )));
204 }
205
206 sqlx::query(
207 "UPDATE project_members SET split_percent = $3 WHERE project_id = $1 AND user_id = $2",
208 )
209 .bind(project_id)
210 .bind(user_id)
211 .bind(new_split_percent)
212 .execute(&mut *tx)
213 .await?;
214
215 tx.commit().await?;
216 Ok(())
217 }
218
219 // ── Revenue Splits ──
220
221 /// Record revenue splits for a completed tip.
222 #[tracing::instrument(skip(pool))]
223 pub(crate) async fn create_tip_splits(
224 pool: &PgPool,
225 tip_id: TipId,
226 splits: &[(UserId, i64, i16)], // (recipient_id, amount_cents, split_percent)
227 ) -> Result<()> {
228 if splits.is_empty() {
229 return Ok(());
230 }
231 let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
232 let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
233 let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
234 // ON CONFLICT DO NOTHING (against the partial revenue_splits_tip_recipient_key,
235 // migration 163): one split row per recipient per tip, so a crash-recovery
236 // redelivery that re-runs `record_tip_splits` is a no-op rather than a
237 // double-credit (Run 20 Payments chronic). The WHERE predicate matches the
238 // partial index so PostgreSQL infers it as the arbiter.
239 sqlx::query(
240 r"
241 INSERT INTO revenue_splits (tip_id, recipient_id, amount_cents, split_percent, status)
242 SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending'
243 ON CONFLICT (tip_id, recipient_id) WHERE tip_id IS NOT NULL DO NOTHING
244 ",
245 )
246 .bind(tip_id)
247 .bind(&recipient_ids)
248 .bind(&amounts)
249 .bind(&percents)
250 .execute(pool)
251 .await?;
252 Ok(())
253 }
254
255 /// Record revenue splits for a completed transaction (item purchase).
256 #[tracing::instrument(skip(pool))]
257 pub(crate) async fn create_transaction_splits(
258 pool: &PgPool,
259 transaction_id: TransactionId,
260 splits: &[(UserId, i64, i16)], // (recipient_id, amount_cents, split_percent)
261 ) -> Result<()> {
262 if splits.is_empty() {
263 return Ok(());
264 }
265 let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
266 let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
267 let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
268 // ON CONFLICT DO NOTHING (against revenue_splits_tx_recipient_key): a
269 // crash-recovery finalize re-run records no duplicate splits.
270 sqlx::query(
271 r"
272 INSERT INTO revenue_splits (transaction_id, recipient_id, amount_cents, split_percent, status)
273 SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending'
274 ON CONFLICT (transaction_id, recipient_id) DO NOTHING
275 ",
276 )
277 .bind(transaction_id)
278 .bind(&recipient_ids)
279 .bind(&amounts)
280 .bind(&percents)
281 .execute(pool)
282 .await?;
283 Ok(())
284 }
285
286 /// Get all revenue splits for a recipient, most recent first.
287 #[allow(dead_code)]
288 #[tracing::instrument(skip(pool))]
289 pub(crate) async fn get_splits_for_recipient(
290 pool: &PgPool,
291 recipient_id: UserId,
292 limit: i64,
293 offset: i64,
294 ) -> Result<Vec<DbRevenueSplit>> {
295 let splits = sqlx::query_as::<_, DbRevenueSplit>(
296 r"
297 SELECT * FROM revenue_splits
298 WHERE recipient_id = $1
299 ORDER BY created_at DESC
300 LIMIT $2 OFFSET $3
301 ",
302 )
303 .bind(recipient_id)
304 .bind(limit)
305 .bind(offset)
306 .fetch_all(pool)
307 .await?;
308
309 Ok(splits)
310 }
311
312 /// Total split revenue owed to a recipient (all completed splits).
313 #[tracing::instrument(skip(pool))]
314 pub(crate) async fn total_split_revenue(pool: &PgPool, recipient_id: UserId) -> Result<i64> {
315 let row: (Option<i64>,) = sqlx::query_as(
316 "SELECT SUM(amount_cents)::BIGINT FROM revenue_splits WHERE recipient_id = $1",
317 )
318 .bind(recipient_id)
319 .fetch_one(pool)
320 .await?;
321
322 Ok(row.0.unwrap_or(0))
323 }
324
325 /// Count of split records for a recipient.
326 #[tracing::instrument(skip(pool))]
327 pub(crate) async fn count_splits_for_recipient(pool: &PgPool, recipient_id: UserId) -> Result<i64> {
328 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM revenue_splits WHERE recipient_id = $1")
329 .bind(recipient_id)
330 .fetch_one(pool)
331 .await?;
332
333 Ok(row.0)
334 }
335
336 /// Get all splits involving a user (as owner or recipient) for CSV export.
337 /// Returns splits where the user is either:
338 /// - The recipient (collaborator receiving a share), or
339 /// - The seller/tip recipient (owner who owes collaborators)
340 #[tracing::instrument(skip(pool))]
341 /// One page of a user's revenue splits for CSV export, newest first.
342 ///
343 /// Paginated so the splits export streams in bounded batches instead of loading
344 /// every split in one query (ultra-fuzz Run 4 S1). Stable `(created_at, id)`
345 /// ordering keeps OFFSET batches consistent.
346 pub(crate) async fn get_splits_for_export_page(
347 pool: &PgPool,
348 user_id: UserId,
349 limit: i64,
350 offset: i64,
351 ) -> Result<Vec<DbSplitExportRow>> {
352 let rows = sqlx::query_as::<_, DbSplitExportRow>(
353 r"
354 SELECT rs.id, rs.recipient_id, rs.amount_cents, rs.split_percent, rs.created_at,
355 CASE WHEN rs.transaction_id IS NOT NULL THEN 'sale' ELSE 'tip' END AS source_type,
356 u.username AS recipient_username
357 FROM revenue_splits rs
358 JOIN users u ON u.id = rs.recipient_id
359 LEFT JOIN transactions t ON t.id = rs.transaction_id
360 LEFT JOIN tips tip ON tip.id = rs.tip_id
361 WHERE rs.recipient_id = $1
362 OR COALESCE(t.seller_id, tip.recipient_id) = $1
363 ORDER BY rs.created_at DESC, rs.id DESC
364 LIMIT $2 OFFSET $3
365 ",
366 )
367 .bind(user_id)
368 .bind(limit)
369 .bind(offset)
370 .fetch_all(pool)
371 .await?;
372
373 Ok(rows)
374 }
375
376 /// Total split obligations owed by a project owner (splits on their transactions/tips).
377 #[tracing::instrument(skip(pool))]
378 pub(crate) async fn total_split_obligations(pool: &PgPool, owner_id: UserId) -> Result<i64> {
379 let row: (Option<i64>,) = sqlx::query_as(
380 r"
381 SELECT SUM(rs.amount_cents)::BIGINT
382 FROM revenue_splits rs
383 LEFT JOIN transactions t ON t.id = rs.transaction_id
384 LEFT JOIN tips tip ON tip.id = rs.tip_id
385 WHERE COALESCE(t.seller_id, tip.recipient_id) = $1
386 ",
387 )
388 .bind(owner_id)
389 .fetch_one(pool)
390 .await?;
391
392 Ok(row.0.unwrap_or(0))
393 }
394