Skip to main content

max / makenotwork

16.8 KB · 493 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 pm.accepted_at,
134 u.username, u.display_name, u.stripe_account_id, u.stripe_charges_enabled,
135 u.settlement_currency
136 FROM project_members pm
137 JOIN users u ON u.id = pm.user_id
138 WHERE pm.project_id = $1
139 ORDER BY pm.split_percent DESC
140 ",
141 )
142 .bind(project_id)
143 .fetch_all(pool)
144 .await?;
145
146 Ok(members)
147 }
148
149 /// Accept a split invitation. Returns false if there was nothing pending.
150 ///
151 /// Idempotent by the `accepted_at IS NULL` guard: a double-click cannot move an
152 /// acceptance timestamp forward, which matters because that timestamp is the
153 /// line between sales the owner kept and sales this collaborator shares in.
154 #[tracing::instrument(skip(pool))]
155 pub(crate) async fn accept_split_invitation(
156 pool: &PgPool,
157 project_id: ProjectId,
158 user_id: UserId,
159 ) -> Result<bool> {
160 let result = sqlx::query(
161 "UPDATE project_members SET accepted_at = NOW() \
162 WHERE project_id = $1 AND user_id = $2 AND accepted_at IS NULL",
163 )
164 .bind(project_id)
165 .bind(user_id)
166 .execute(pool)
167 .await?;
168 Ok(result.rows_affected() > 0)
169 }
170
171 /// Decline a split invitation, removing the membership and freeing the reserved
172 /// percentage. Only ever touches a row that is still pending, so it cannot undo
173 /// an acceptance.
174 #[tracing::instrument(skip(pool))]
175 pub(crate) async fn decline_split_invitation(
176 pool: &PgPool,
177 project_id: ProjectId,
178 user_id: UserId,
179 ) -> Result<bool> {
180 let result = sqlx::query(
181 "DELETE FROM project_members \
182 WHERE project_id = $1 AND user_id = $2 AND accepted_at IS NULL",
183 )
184 .bind(project_id)
185 .bind(user_id)
186 .execute(pool)
187 .await?;
188 Ok(result.rows_affected() > 0)
189 }
190
191 /// A split invitation awaiting this creator's answer.
192 #[derive(Debug, Clone, sqlx::FromRow)]
193 pub struct PendingSplitInvitation {
194 pub project_id: ProjectId,
195 pub project_title: String,
196 pub project_slug: crate::db::Slug,
197 pub owner_username: String,
198 pub split_percent: i16,
199 pub added_at: chrono::DateTime<chrono::Utc>,
200 /// The currency the project sells in, which is its owner's. If it differs
201 /// from the invited creator's own, their share converts at their payout and
202 /// they carry that cost, which is the whole reason this screen exists.
203 pub project_currency: crate::currency::SettlementCurrency,
204 }
205
206 /// Every split invitation this creator has not yet answered.
207 #[tracing::instrument(skip(pool))]
208 pub(crate) async fn get_pending_invitations(
209 pool: &PgPool,
210 user_id: UserId,
211 ) -> Result<Vec<PendingSplitInvitation>> {
212 let rows = sqlx::query_as::<_, PendingSplitInvitation>(
213 r"
214 SELECT pm.project_id, p.title AS project_title, p.slug AS project_slug,
215 owner.username AS owner_username, pm.split_percent, pm.added_at,
216 owner.settlement_currency AS project_currency
217 FROM project_members pm
218 JOIN projects p ON p.id = pm.project_id
219 JOIN users owner ON owner.id = p.user_id
220 WHERE pm.user_id = $1 AND pm.accepted_at IS NULL
221 ORDER BY pm.added_at DESC
222 LIMIT 100
223 ",
224 )
225 .bind(user_id)
226 .fetch_all(pool)
227 .await?;
228 Ok(rows)
229 }
230
231 /// Get the total split percentage allocated to members (excluding the owner).
232 ///
233 /// Counts pending invitations too. The percentage is reserved the moment it is
234 /// offered, so an owner cannot promise the same revenue twice while the first
235 /// collaborator has not answered.
236 #[tracing::instrument(skip(pool))]
237 pub(crate) async fn get_total_split_percent(pool: &PgPool, project_id: ProjectId) -> Result<i64> {
238 let row: (Option<i64>,) = sqlx::query_as(
239 "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
240 )
241 .bind(project_id)
242 .fetch_one(pool)
243 .await?;
244
245 Ok(row.0.unwrap_or(0))
246 }
247
248 /// Update a member's split percentage.
249 #[allow(dead_code)]
250 #[tracing::instrument(skip(pool))]
251 pub(crate) async fn update_member_split(
252 pool: &PgPool,
253 project_id: ProjectId,
254 user_id: UserId,
255 new_split_percent: i16,
256 ) -> Result<()> {
257 if !(0..=100).contains(&new_split_percent) {
258 return Err(AppError::BadRequest(format!(
259 "split_percent must be between 0 and 100 (got {new_split_percent})"
260 )));
261 }
262 let mut tx = pool.begin().await?;
263
264 // Serialize on the parent project row (see `add_project_member`). The prior
265 // `SUM(...) FOR UPDATE` below was invalid, Postgres rejects FOR UPDATE with
266 // an aggregate, so this function 500'd on every call (audit Run 13); the lock
267 // now lives on the projects row and the SUM runs plain.
268 lock_project_for_splits(&mut tx, project_id).await?;
269
270 let current: Option<DbProjectMember> =
271 sqlx::query_as("SELECT * FROM project_members WHERE project_id = $1 AND user_id = $2")
272 .bind(project_id)
273 .bind(user_id)
274 .fetch_optional(&mut *tx)
275 .await?;
276
277 let current_member_split = current.map_or(0, |m| m.split_percent as i64);
278
279 let total_row: (Option<i64>,) = sqlx::query_as(
280 "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
281 )
282 .bind(project_id)
283 .fetch_one(&mut *tx)
284 .await?;
285
286 let total = total_row.0.unwrap_or(0);
287 let new_total = total - current_member_split + new_split_percent as i64;
288 if new_total > 100 {
289 return Err(AppError::BadRequest(format!(
290 "Total split would be {new_total}%, exceeding 100%"
291 )));
292 }
293
294 sqlx::query(
295 "UPDATE project_members SET split_percent = $3 WHERE project_id = $1 AND user_id = $2",
296 )
297 .bind(project_id)
298 .bind(user_id)
299 .bind(new_split_percent)
300 .execute(&mut *tx)
301 .await?;
302
303 tx.commit().await?;
304 Ok(())
305 }
306
307 // ── Revenue Splits ──
308
309 /// Record revenue splits for a completed tip.
310 #[tracing::instrument(skip(pool))]
311 pub(crate) async fn create_tip_splits(
312 pool: &PgPool,
313 tip_id: TipId,
314 splits: &[(UserId, i64, i16)], // (recipient_id, amount_cents, split_percent)
315 ) -> Result<()> {
316 if splits.is_empty() {
317 return Ok(());
318 }
319 let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
320 let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
321 let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
322 // ON CONFLICT DO NOTHING (against the partial revenue_splits_tip_recipient_key,
323 // migration 163): one split row per recipient per tip, so a crash-recovery
324 // redelivery that re-runs `record_tip_splits` is a no-op rather than a
325 // double-credit (Run 20 Payments chronic). The WHERE predicate matches the
326 // partial index so PostgreSQL infers it as the arbiter.
327 sqlx::query(
328 r"
329 INSERT INTO revenue_splits (tip_id, recipient_id, amount_cents, split_percent, status, currency)
330 SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending',
331 (SELECT currency FROM tips WHERE id = $1)
332 ON CONFLICT (tip_id, recipient_id) WHERE tip_id IS NOT NULL DO NOTHING
333 ",
334 )
335 .bind(tip_id)
336 .bind(&recipient_ids)
337 .bind(&amounts)
338 .bind(&percents)
339 .execute(pool)
340 .await?;
341 Ok(())
342 }
343
344 /// Record revenue splits for a completed transaction (item purchase).
345 #[tracing::instrument(skip(pool))]
346 pub(crate) async fn create_transaction_splits(
347 pool: &PgPool,
348 transaction_id: TransactionId,
349 splits: &[(UserId, i64, i16)], // (recipient_id, amount_cents, split_percent)
350 ) -> Result<()> {
351 if splits.is_empty() {
352 return Ok(());
353 }
354 let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
355 let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
356 let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
357 // ON CONFLICT DO NOTHING (against revenue_splits_tx_recipient_key): a
358 // crash-recovery finalize re-run records no duplicate splits.
359 sqlx::query(
360 r"
361 INSERT INTO revenue_splits (transaction_id, recipient_id, amount_cents, split_percent, status, currency)
362 SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending',
363 (SELECT currency FROM transactions WHERE id = $1)
364 ON CONFLICT (transaction_id, recipient_id) DO NOTHING
365 ",
366 )
367 .bind(transaction_id)
368 .bind(&recipient_ids)
369 .bind(&amounts)
370 .bind(&percents)
371 .execute(pool)
372 .await?;
373 Ok(())
374 }
375
376 /// Get all revenue splits for a recipient, most recent first.
377 #[allow(dead_code)]
378 #[tracing::instrument(skip(pool))]
379 pub(crate) async fn get_splits_for_recipient(
380 pool: &PgPool,
381 recipient_id: UserId,
382 limit: i64,
383 offset: i64,
384 ) -> Result<Vec<DbRevenueSplit>> {
385 let splits = sqlx::query_as::<_, DbRevenueSplit>(
386 r"
387 SELECT * FROM revenue_splits
388 WHERE recipient_id = $1
389 ORDER BY created_at DESC
390 LIMIT $2 OFFSET $3
391 ",
392 )
393 .bind(recipient_id)
394 .bind(limit)
395 .bind(offset)
396 .fetch_all(pool)
397 .await?;
398
399 Ok(splits)
400 }
401
402 /// Total split revenue owed to a recipient (all completed splits).
403 #[tracing::instrument(skip(pool))]
404 pub(crate) async fn total_split_revenue(
405 pool: &PgPool,
406 recipient_id: UserId,
407 ) -> Result<crate::currency::MoneyByCurrency> {
408 // Grouped, because this is the one place a creator genuinely holds money in
409 // someone else's currency: a split is denominated in the paying project's
410 // currency, which is its owner's, not the recipient's.
411 let rows: Vec<(crate::currency::SettlementCurrency, Option<i64>)> = sqlx::query_as(
412 "SELECT currency, SUM(amount_cents)::BIGINT FROM revenue_splits \
413 WHERE recipient_id = $1 GROUP BY currency",
414 )
415 .bind(recipient_id)
416 .fetch_all(pool)
417 .await?;
418
419 Ok(crate::currency::MoneyByCurrency::from_rows(
420 rows.into_iter().map(|(c, cents)| (c, cents.unwrap_or(0))),
421 ))
422 }
423
424 /// Count of split records for a recipient.
425 #[tracing::instrument(skip(pool))]
426 pub(crate) async fn count_splits_for_recipient(pool: &PgPool, recipient_id: UserId) -> Result<i64> {
427 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM revenue_splits WHERE recipient_id = $1")
428 .bind(recipient_id)
429 .fetch_one(pool)
430 .await?;
431
432 Ok(row.0)
433 }
434
435 /// Get all splits involving a user (as owner or recipient) for CSV export.
436 /// Returns splits where the user is either:
437 /// - The recipient (collaborator receiving a share), or
438 /// - The seller/tip recipient (owner who owes collaborators)
439 #[tracing::instrument(skip(pool))]
440 /// One page of a user's revenue splits for CSV export, newest first.
441 ///
442 /// Paginated so the splits export streams in bounded batches instead of loading
443 /// every split in one query (ultra-fuzz Run 4 S1). Stable `(created_at, id)`
444 /// ordering keeps OFFSET batches consistent.
445 pub(crate) async fn get_splits_for_export_page(
446 pool: &PgPool,
447 user_id: UserId,
448 limit: i64,
449 offset: i64,
450 ) -> Result<Vec<DbSplitExportRow>> {
451 let rows = sqlx::query_as::<_, DbSplitExportRow>(
452 r"
453 SELECT rs.id, rs.recipient_id, rs.amount_cents, rs.split_percent, rs.created_at,
454 CASE WHEN rs.transaction_id IS NOT NULL THEN 'sale' ELSE 'tip' END AS source_type,
455 u.username AS recipient_username
456 FROM revenue_splits rs
457 JOIN users u ON u.id = rs.recipient_id
458 LEFT JOIN transactions t ON t.id = rs.transaction_id
459 LEFT JOIN tips tip ON tip.id = rs.tip_id
460 WHERE rs.recipient_id = $1
461 OR COALESCE(t.seller_id, tip.recipient_id) = $1
462 ORDER BY rs.created_at DESC, rs.id DESC
463 LIMIT $2 OFFSET $3
464 ",
465 )
466 .bind(user_id)
467 .bind(limit)
468 .bind(offset)
469 .fetch_all(pool)
470 .await?;
471
472 Ok(rows)
473 }
474
475 /// Total split obligations owed by a project owner (splits on their transactions/tips).
476 #[tracing::instrument(skip(pool))]
477 pub(crate) async fn total_split_obligations(pool: &PgPool, owner_id: UserId) -> Result<i64> {
478 let row: (Option<i64>,) = sqlx::query_as(
479 r"
480 SELECT SUM(rs.amount_cents)::BIGINT
481 FROM revenue_splits rs
482 LEFT JOIN transactions t ON t.id = rs.transaction_id
483 LEFT JOIN tips tip ON tip.id = rs.tip_id
484 WHERE COALESCE(t.seller_id, tip.recipient_id) = $1
485 ",
486 )
487 .bind(owner_id)
488 .fetch_one(pool)
489 .await?;
490
491 Ok(row.0.unwrap_or(0))
492 }
493