| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 5 |
|
| 6 |
|
| 7 |
use sqlx::PgPool; |
| 8 |
|
| 9 |
use crate::db::UserId; |
| 10 |
use crate::db::validated_types::Email; |
| 11 |
use crate::error::Result; |
| 12 |
|
| 13 |
|
| 14 |
|
| 15 |
|
| 16 |
|
| 17 |
|
| 18 |
#[tracing::instrument(skip_all)] |
| 19 |
pub async fn get_all_user_emails(pool: &PgPool) -> Result<Vec<(String, Option<String>)>> { |
| 20 |
const ALL_EMAILS_CAP: i64 = 50_000; |
| 21 |
let rows = sqlx::query_as::<_, (String, Option<String>)>( |
| 22 |
"SELECT email, display_name FROM users ORDER BY created_at ASC LIMIT $1", |
| 23 |
) |
| 24 |
.bind(ALL_EMAILS_CAP) |
| 25 |
.fetch_all(pool) |
| 26 |
.await?; |
| 27 |
|
| 28 |
if rows.len() as i64 == ALL_EMAILS_CAP { |
| 29 |
tracing::warn!( |
| 30 |
cap = ALL_EMAILS_CAP, |
| 31 |
"get_all_user_emails hit its cap; some users omitted, switch to paged dispatch" |
| 32 |
); |
| 33 |
} |
| 34 |
|
| 35 |
Ok(rows) |
| 36 |
} |
| 37 |
|
| 38 |
|
| 39 |
|
| 40 |
|
| 41 |
|
| 42 |
|
| 43 |
#[derive(Debug, Clone, Copy)] |
| 44 |
pub struct NotificationPreferences { |
| 45 |
pub notify_sale: bool, |
| 46 |
pub notify_follower: bool, |
| 47 |
pub notify_release: bool, |
| 48 |
pub login_notification_enabled: bool, |
| 49 |
pub notify_issues: bool, |
| 50 |
pub notify_status: bool, |
| 51 |
pub notify_invite: bool, |
| 52 |
} |
| 53 |
|
| 54 |
#[tracing::instrument(skip_all)] |
| 55 |
pub async fn update_notification_preferences( |
| 56 |
pool: &PgPool, |
| 57 |
id: UserId, |
| 58 |
prefs: NotificationPreferences, |
| 59 |
) -> Result<()> { |
| 60 |
let NotificationPreferences { |
| 61 |
notify_sale, |
| 62 |
notify_follower, |
| 63 |
notify_release, |
| 64 |
login_notification_enabled, |
| 65 |
notify_issues, |
| 66 |
notify_status, |
| 67 |
notify_invite, |
| 68 |
} = prefs; |
| 69 |
|
| 70 |
for (kind, enabled) in [ |
| 71 |
("sale", notify_sale), |
| 72 |
("follower", notify_follower), |
| 73 |
("releases", notify_release), |
| 74 |
("issues", notify_issues), |
| 75 |
("status", notify_status), |
| 76 |
("login", login_notification_enabled), |
| 77 |
("invite", notify_invite), |
| 78 |
] { |
| 79 |
crate::db::lists::sync_notification_subscription(pool, id, kind, enabled).await?; |
| 80 |
} |
| 81 |
|
| 82 |
Ok(()) |
| 83 |
} |
| 84 |
|
| 85 |
|
| 86 |
|
| 87 |
|
| 88 |
|
| 89 |
|
| 90 |
|
| 91 |
#[tracing::instrument(skip_all)] |
| 92 |
pub async fn update_tip_preferences( |
| 93 |
pool: &PgPool, |
| 94 |
id: UserId, |
| 95 |
tips_enabled: bool, |
| 96 |
notify_tip: bool, |
| 97 |
) -> Result<()> { |
| 98 |
sqlx::query("UPDATE users SET tips_enabled = $2, updated_at = NOW() WHERE id = $1") |
| 99 |
.bind(id) |
| 100 |
.bind(tips_enabled) |
| 101 |
.execute(pool) |
| 102 |
.await?; |
| 103 |
|
| 104 |
crate::db::lists::sync_notification_subscription(pool, id, "tip", notify_tip).await?; |
| 105 |
Ok(()) |
| 106 |
} |
| 107 |
|
| 108 |
|
| 109 |
|
| 110 |
|
| 111 |
|
| 112 |
|
| 113 |
|
| 114 |
|
| 115 |
|
| 116 |
|
| 117 |
|
| 118 |
#[tracing::instrument(skip_all)] |
| 119 |
pub async fn disable_notification( |
| 120 |
pool: &PgPool, |
| 121 |
user_id: UserId, |
| 122 |
preference: &str, |
| 123 |
) -> Result<bool> { |
| 124 |
let legacy: Option<&str> = crate::db::lists::NOTIFICATION_LISTS |
| 125 |
.iter() |
| 126 |
.find(|(_, legacy)| *legacy == preference) |
| 127 |
.map(|(kind, _)| *kind); |
| 128 |
let Some(kind) = legacy.or_else(|| { |
| 129 |
preference |
| 130 |
.parse::<crate::db::ListKind>() |
| 131 |
.ok() |
| 132 |
.map(|_| preference) |
| 133 |
}) else { |
| 134 |
return Ok(false); |
| 135 |
}; |
| 136 |
crate::db::lists::sync_notification_subscription(pool, user_id, kind, false).await?; |
| 137 |
Ok(true) |
| 138 |
} |
| 139 |
|
| 140 |
|
| 141 |
#[derive(sqlx::FromRow)] |
| 142 |
pub struct StatusAlertSubscriber { |
| 143 |
pub id: UserId, |
| 144 |
pub email: Email, |
| 145 |
pub display_name: Option<String>, |
| 146 |
} |
| 147 |
|
| 148 |
|
| 149 |
|
| 150 |
|
| 151 |
|
| 152 |
|
| 153 |
|
| 154 |
|
| 155 |
#[tracing::instrument(skip_all)] |
| 156 |
pub async fn get_status_alert_subscribers(pool: &PgPool) -> Result<Vec<StatusAlertSubscriber>> { |
| 157 |
const STATUS_SUBSCRIBER_CAP: i64 = 10_000; |
| 158 |
let rows = sqlx::query_as::<_, StatusAlertSubscriber>( |
| 159 |
"SELECT u.id, u.email, u.display_name FROM users u \ |
| 160 |
JOIN list_subscriptions ls ON ls.user_id = u.id \ |
| 161 |
JOIN lists l ON l.id = ls.list_id AND l.scope = 'platform' AND l.kind = 'status' \ |
| 162 |
WHERE ls.state IN ('confirmed', 'imported') AND u.deactivated_at IS NULL \ |
| 163 |
ORDER BY u.id LIMIT $1", |
| 164 |
) |
| 165 |
.bind(STATUS_SUBSCRIBER_CAP) |
| 166 |
.fetch_all(pool) |
| 167 |
.await?; |
| 168 |
if rows.len() as i64 == STATUS_SUBSCRIBER_CAP { |
| 169 |
tracing::warn!( |
| 170 |
cap = STATUS_SUBSCRIBER_CAP, |
| 171 |
"get_status_alert_subscribers hit hard cap; promote to paged dispatch" |
| 172 |
); |
| 173 |
} |
| 174 |
Ok(rows) |
| 175 |
} |
| 176 |
|
| 177 |
|
| 178 |
#[tracing::instrument(skip_all)] |
| 179 |
pub async fn try_set_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<bool> { |
| 180 |
let result = sqlx::query( |
| 181 |
r" |
| 182 |
UPDATE users |
| 183 |
SET last_broadcast_at = NOW() |
| 184 |
WHERE id = $1 |
| 185 |
AND (last_broadcast_at IS NULL OR last_broadcast_at < NOW() - INTERVAL '24 hours') |
| 186 |
", |
| 187 |
) |
| 188 |
.bind(user_id) |
| 189 |
.execute(pool) |
| 190 |
.await?; |
| 191 |
|
| 192 |
Ok(result.rows_affected() > 0) |
| 193 |
} |
| 194 |
|
| 195 |
|
| 196 |
|
| 197 |
|
| 198 |
#[tracing::instrument(skip_all)] |
| 199 |
pub async fn clear_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<()> { |
| 200 |
sqlx::query("UPDATE users SET last_broadcast_at = NULL WHERE id = $1") |
| 201 |
.bind(user_id) |
| 202 |
.execute(pool) |
| 203 |
.await?; |
| 204 |
Ok(()) |
| 205 |
} |
| 206 |
|