| 1 | |
|
| 2 | |
|
| 3 | |
|
| 4 | | use super::{now_ms, Pool}; |
| 5 | | use sqlx::FromRow; |
| 6 | |
|
| 7 | | #[derive(Clone, Debug, FromRow)] |
| 8 | | pub struct Notification { |
| 9 | | pub id: i64, |
| 10 | | pub account_id: i64, |
| 11 | | pub at_ms: i64, |
| 12 | | pub kind: String, |
| 13 | | pub title: String, |
| 14 | | pub url: String, |
| 15 | | pub actor: String, |
| 16 | | pub read: bool, |
| 17 | | pub repo_id: Option<i64>, |
| 18 | | pub thread: String, |
| 19 | | pub reason: String, |
| 20 | | pub done: bool, |
| 21 | | } |
| 22 | |
|
| 23 | |
|
| 24 | | #[derive(Clone, Copy, Debug, PartialEq, Eq)] |
| 25 | | pub enum Reason { |
| 26 | | Participating, |
| 27 | | Watching, |
| 28 | | } |
| 29 | |
|
| 30 | | impl Reason { |
| 31 | 107× | pub fn as_str(self) -> &'static str { |
| 32 | 107× | match self { |
| 33 | 75× | Reason::Participating => "participating", |
| 34 | 32× | Reason::Watching => "watching", |
| 35 | | } |
| 36 | 107× | } |
| 37 | | } |
| 38 | |
|
| 39 | |
|
| 40 | 214× | pub fn thread_of(url: &str) -> &str { |
| 41 | 214× | url.split('#').next().unwrap_or(url) |
| 42 | 214× | } |
| 43 | |
|
| 44 | |
|
| 45 | | pub struct New<'a> { |
| 46 | | pub account_id: i64, |
| 47 | | pub repo_id: Option<i64>, |
| 48 | | pub reason: Reason, |
| 49 | | pub kind: &'a str, |
| 50 | | pub title: &'a str, |
| 51 | | pub url: &'a str, |
| 52 | | pub actor: &'a str, |
| 53 | | } |
| 54 | |
|
| 55 | 107× | pub async fn add(pool: &Pool, n: New<'_>) -> sqlx::Result<()> { |
| 56 | 107× | let New { account_id, repo_id, reason, kind, title, url, actor } = n; |
| 57 | | |
| 58 | 107× | sqlx::query("UPDATE notifications SET done = FALSE WHERE account_id = $1 AND thread = $2 AND done = TRUE") |
| 59 | 107× | .bind(account_id) |
| 60 | 107× | .bind(thread_of(url)) |
| 61 | 107× | .execute(pool) |
| 62 | 107× | .await?; |
| 63 | 107× | sqlx::query("INSERT INTO notifications (account_id, at_ms, kind, title, url, actor, repo_id, thread, reason) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)") |
| 64 | 107× | .bind(account_id) |
| 65 | 107× | .bind(now_ms()) |
| 66 | 107× | .bind(kind) |
| 67 | 107× | .bind(title) |
| 68 | 107× | .bind(url) |
| 69 | 107× | .bind(actor) |
| 70 | 107× | .bind(repo_id) |
| 71 | 107× | .bind(thread_of(url)) |
| 72 | 107× | .bind(reason.as_str()) |
| 73 | 107× | .execute(pool) |
| 74 | 107× | .await?; |
| 75 | 107× | Ok(()) |
| 76 | 107× | } |
| 77 | |
|
| 78 | |
|
| 79 | | #[derive(Clone, Debug, FromRow)] |
| 80 | | pub struct Thread { |
| 81 | | |
| 82 | | pub id: i64, |
| 83 | | pub at_ms: i64, |
| 84 | | pub kind: String, |
| 85 | | pub title: String, |
| 86 | | pub url: String, |
| 87 | | pub actor: String, |
| 88 | | pub thread: String, |
| 89 | | pub repo_id: Option<i64>, |
| 90 | | pub repo: Option<String>, |
| 91 | | pub count: i64, |
| 92 | | pub unread: i64, |
| 93 | | pub participating: bool, |
| 94 | | pub done: bool, |
| 95 | | } |
| 96 | |
|
| 97 | |
|
| 98 | | #[derive(Clone, Debug, Default)] |
| 99 | | pub struct Filter { |
| 100 | | pub done: bool, |
| 101 | | pub unread_only: bool, |
| 102 | | |
| 103 | | pub reason: Option<String>, |
| 104 | | |
| 105 | | pub kind: Option<String>, |
| 106 | | pub repo_id: Option<i64>, |
| 107 | | } |
| 108 | |
|
| 109 | 36× | pub async fn threads(pool: &Pool, account_id: i64, f: &Filter, limit: i64) -> sqlx::Result<Vec<Thread>> { |
| 110 | | |
| 111 | | |
| 112 | | |
| 113 | 36× | let mut sql = String::from( |
| 114 | | "SELECT id, at_ms, kind, title, url, actor, thread, repo_id, \ |
| 115 | | (SELECT r.owner || '/' || r.name FROM repositories r WHERE r.id = t.repo_id) AS repo, \ |
| 116 | | count, unread, participating > 0 AS participating, done FROM (\ |
| 117 | | SELECT n.id, n.at_ms, n.kind, n.title, n.url, n.actor, n.thread, n.repo_id, n.done, \ |
| 118 | | ROW_NUMBER() OVER (PARTITION BY n.thread ORDER BY n.at_ms DESC, n.id DESC) AS newest, \ |
| 119 | | COUNT(*) OVER (PARTITION BY n.thread) AS count, \ |
| 120 | | SUM(CASE WHEN n.read THEN 0 ELSE 1 END) OVER (PARTITION BY n.thread) AS unread, \ |
| 121 | | SUM(CASE WHEN n.reason = 'participating' THEN 1 ELSE 0 END) OVER (PARTITION BY n.thread) AS participating \ |
| 122 | | FROM notifications n WHERE n.account_id = $1 AND n.done = $2", |
| 123 | | ); |
| 124 | | |
| 125 | 36× | let mut next = 3; |
| 126 | 36× | if f.kind.is_some() { |
| 127 | 1× | sql.push_str(&format!(" AND (n.kind = ${} OR n.kind LIKE ${} || '.%')", next, next + 1)); |
| 128 | 1× | next += 2; |
| 129 | 35× | } |
| 130 | 36× | if f.repo_id.is_some() { |
| 131 | 2× | sql.push_str(&format!(" AND n.repo_id = ${next}")); |
| 132 | 2× | next += 1; |
| 133 | 34× | } |
| 134 | 36× | sql.push_str(") t WHERE newest = 1"); |
| 135 | 36× | if f.unread_only { |
| 136 | 1× | sql.push_str(" AND unread > 0"); |
| 137 | 35× | } |
| 138 | 36× | match f.reason.as_deref() { |
| 139 | 2× | Some("participating") => sql.push_str(" AND participating > 0"), |
| 140 | 1× | Some("watching") => sql.push_str(" AND participating = 0"), |
| 141 | 34× | _ => {} |
| 142 | | } |
| 143 | 36× | sql.push_str(&format!(" ORDER BY at_ms DESC LIMIT ${next}")); |
| 144 | 36× | let mut q = sqlx::query_as::<_, Thread>(&sql).bind(account_id).bind(f.done); |
| 145 | 36× | if let Some(k) = &f.kind { |
| 146 | 1× | q = q.bind(k).bind(k); |
| 147 | 35× | } |
| 148 | 36× | if let Some(r) = f.repo_id { |
| 149 | 2× | q = q.bind(r); |
| 150 | 34× | } |
| 151 | 36× | q.bind(limit).fetch_all(pool).await |
| 152 | 36× | } |
| 153 | |
|
| 154 | |
|
| 155 | 3× | pub async fn set_thread(pool: &Pool, account_id: i64, thread: &str, read: Option<bool>, done: Option<bool>) -> sqlx::Result<()> { |
| 156 | 3× | if let Some(r) = read { |
| 157 | 2× | if r { |
| 158 | 2× | sqlx::query("UPDATE notifications SET read = TRUE WHERE account_id = $1 AND thread = $2") |
| 159 | 2× | .bind(account_id) |
| 160 | 2× | .bind(thread) |
| 161 | 2× | .execute(pool) |
| 162 | 2× | .await?; |
| 163 | | } else { |
| 164 | | |
| 165 | 0× | sqlx::query("UPDATE notifications SET read = FALSE WHERE id = (SELECT id FROM notifications WHERE account_id = $1 AND thread = $2 ORDER BY at_ms DESC, id DESC LIMIT 1)") |
| 166 | 0× | .bind(account_id) |
| 167 | 0× | .bind(thread) |
| 168 | 0× | .execute(pool) |
| 169 | 0× | .await?; |
| 170 | | } |
| 171 | 1× | } |
| 172 | 3× | if let Some(d) = done { |
| 173 | 1× | let read_too = if d { ", read = TRUE" } else { "" }; |
| 174 | 1× | sqlx::query(&format!("UPDATE notifications SET done = $1{read_too} WHERE account_id = $2 AND thread = $3")) |
| 175 | 1× | .bind(d) |
| 176 | 1× | .bind(account_id) |
| 177 | 1× | .bind(thread) |
| 178 | 1× | .execute(pool) |
| 179 | 1× | .await?; |
| 180 | 2× | } |
| 181 | 3× | Ok(()) |
| 182 | 3× | } |
| 183 | |
|
| 184 | |
|
| 185 | 2× | pub async fn get(pool: &Pool, account_id: i64, id: i64) -> sqlx::Result<Option<Notification>> { |
| 186 | 2× | sqlx::query_as("SELECT * FROM notifications WHERE account_id = $1 AND id = $2").bind(account_id).bind(id).fetch_optional(pool).await |
| 187 | 2× | } |
| 188 | |
|
| 189 | 23× | pub async fn inbox(pool: &Pool, account_id: i64, limit: i64) -> sqlx::Result<Vec<Notification>> { |
| 190 | 23× | sqlx::query_as("SELECT * FROM notifications WHERE account_id = $1 ORDER BY read, at_ms DESC, id DESC LIMIT $2") |
| 191 | 23× | .bind(account_id) |
| 192 | 23× | .bind(limit) |
| 193 | 23× | .fetch_all(pool) |
| 194 | 23× | .await |
| 195 | 23× | } |
| 196 | |
|
| 197 | 1137× | pub async fn unread(pool: &Pool, account_id: i64) -> sqlx::Result<i64> { |
| 198 | 1137× | sqlx::query_scalar("SELECT COUNT(*) FROM notifications WHERE account_id = $1 AND read = FALSE AND done = FALSE") |
| 199 | 1137× | .bind(account_id) |
| 200 | 1137× | .fetch_one(pool) |
| 201 | 1137× | .await |
| 202 | 1137× | } |
| 203 | |
|
| 204 | 5× | pub async fn mark_read(pool: &Pool, account_id: i64, id: Option<i64>) -> sqlx::Result<()> { |
| 205 | 5× | match id { |
| 206 | 1× | Some(id) => { |
| 207 | 1× | sqlx::query("UPDATE notifications SET read = TRUE WHERE account_id = $1 AND id = $2") |
| 208 | 1× | .bind(account_id) |
| 209 | 1× | .bind(id) |
| 210 | 1× | .execute(pool) |
| 211 | 1× | .await? |
| 212 | | } |
| 213 | 4× | None => sqlx::query("UPDATE notifications SET read = TRUE WHERE account_id = $1").bind(account_id).execute(pool).await?, |
| 214 | | }; |
| 215 | 5× | Ok(()) |
| 216 | 5× | } |
| 217 | |
|
| 218 | |
|
| 219 | |
|
| 220 | |
|
| 221 | |
|
| 222 | |
|
| 223 | 225× | pub async fn owners_of(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<i64>> { |
| 224 | 225× | let owner: Option<String> = |
| 225 | 225× | sqlx::query_scalar("SELECT owner FROM repositories WHERE id = $1").bind(repo_id).fetch_optional(pool).await?; |
| 226 | 225× | let Some(owner) = owner else { return Ok(Vec::new()) }; |
| 227 | 224× | if let Some(a) = super::accounts::account_by_name(pool, &owner).await? { |
| 228 | 204× | return Ok(vec![a.id]); |
| 229 | 20× | } |
| 230 | 20× | match super::accounts::org_by_name(pool, &owner).await? { |
| 231 | 4× | Some(o) => { |
| 232 | 6× | Ok(super::accounts::org_members(pool, o.id).await?.into_iter().filter(|(_, r)| r == "owner").map(|(a, _)| a.id).collect()) |
| 233 | | } |
| 234 | 16× | None => Ok(Vec::new()), |
| 235 | | } |
| 236 | 225× | } |
| 237 | |
|
| 238 | |
|
| 239 | |
|
| 240 | 30× | pub async fn watching(pool: &Pool, account_id: i64, repo_id: i64) -> sqlx::Result<bool> { |
| 241 | 30× | let muted: Option<bool> = sqlx::query_scalar("SELECT muted FROM watches WHERE account_id = $1 AND repo_id = $2") |
| 242 | 30× | .bind(account_id) |
| 243 | 30× | .bind(repo_id) |
| 244 | 30× | .fetch_optional(pool) |
| 245 | 30× | .await?; |
| 246 | 30× | match muted { |
| 247 | 4× | Some(m) => Ok(!m), |
| 248 | 26× | None => Ok(owners_of(pool, repo_id).await?.contains(&account_id)), |
| 249 | | } |
| 250 | 30× | } |
| 251 | |
|
| 252 | |
|
| 253 | 5× | pub async fn set_watch(pool: &Pool, account_id: i64, repo_id: i64, on: bool) -> sqlx::Result<()> { |
| 254 | 5× | sqlx::query("INSERT INTO watches (account_id, repo_id, muted) VALUES ($1, $2, $3) ON CONFLICT (account_id, repo_id) DO UPDATE SET muted = excluded.muted") |
| 255 | 5× | .bind(account_id) |
| 256 | 5× | .bind(repo_id) |
| 257 | 5× | .bind(!on) |
| 258 | 5× | .execute(pool) |
| 259 | 5× | .await?; |
| 260 | 5× | Ok(()) |
| 261 | 5× | } |
| 262 | |
|
| 263 | |
|
| 264 | |
|
| 265 | 199× | pub async fn watchers(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<i64>> { |
| 266 | 199× | let rows: Vec<(i64, bool)> = |
| 267 | 199× | sqlx::query_as("SELECT account_id, muted FROM watches WHERE repo_id = $1").bind(repo_id).fetch_all(pool).await?; |
| 268 | 199× | let mut out: Vec<i64> = rows.iter().filter(|(_, m)| !m).map(|(a, _)| *a).collect(); |
| 269 | 199× | for o in owners_of(pool, repo_id).await? { |
| 270 | 182× | if !rows.iter().any(|(a, _)| *a == o) && !out.contains(&o) { |
| 271 | 177× | out.push(o); |
| 272 | 177× | } |
| 273 | | } |
| 274 | 199× | Ok(out) |
| 275 | 199× | } |
| 276 | |
|
| 277 | 100× | pub async fn subscribe(pool: &Pool, account_id: i64, kind: &str, target_id: i64) -> sqlx::Result<()> { |
| 278 | 100× | sqlx::query("INSERT INTO subscriptions (account_id, kind, target_id) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING") |
| 279 | 100× | .bind(account_id) |
| 280 | 100× | .bind(kind) |
| 281 | 100× | .bind(target_id) |
| 282 | 100× | .execute(pool) |
| 283 | 100× | .await?; |
| 284 | 100× | Ok(()) |
| 285 | 100× | } |
| 286 | |
|
| 287 | 0× | pub async fn unsubscribe(pool: &Pool, account_id: i64, kind: &str, target_id: i64) -> sqlx::Result<()> { |
| 288 | 0× | sqlx::query("DELETE FROM subscriptions WHERE account_id = $1 AND kind = $2 AND target_id = $3") |
| 289 | 0× | .bind(account_id) |
| 290 | 0× | .bind(kind) |
| 291 | 0× | .bind(target_id) |
| 292 | 0× | .execute(pool) |
| 293 | 0× | .await?; |
| 294 | 0× | Ok(()) |
| 295 | 0× | } |
| 296 | |
|
| 297 | 17× | pub async fn subscribed(pool: &Pool, account_id: i64, kind: &str, target_id: i64) -> sqlx::Result<bool> { |
| 298 | 17× | let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM subscriptions WHERE account_id = $1 AND kind = $2 AND target_id = $3") |
| 299 | 17× | .bind(account_id) |
| 300 | 17× | .bind(kind) |
| 301 | 17× | .bind(target_id) |
| 302 | 17× | .fetch_one(pool) |
| 303 | 17× | .await?; |
| 304 | 17× | Ok(n > 0) |
| 305 | 17× | } |
| 306 | |
|
| 307 | 102× | pub async fn subscribers(pool: &Pool, kind: &str, target_id: i64) -> sqlx::Result<Vec<i64>> { |
| 308 | 102× | sqlx::query_scalar("SELECT account_id FROM subscriptions WHERE kind = $1 AND target_id = $2") |
| 309 | 102× | .bind(kind) |
| 310 | 102× | .bind(target_id) |
| 311 | 102× | .fetch_all(pool) |
| 312 | 102× | .await |
| 313 | 102× | } |
| 314 | |
|
| 315 | |
|
| 316 | |
|
| 317 | | #[derive(Clone, Debug, FromRow)] |
| 318 | | pub struct Hit { |
| 319 | | pub kind: String, |
| 320 | | pub repo_id: i64, |
| 321 | | pub target_id: i64, |
| 322 | | pub number: i64, |
| 323 | | pub title: String, |
| 324 | | pub snippet: String, |
| 325 | | } |
| 326 | |
|
| 327 | |
|
| 328 | 76× | pub async fn index(pool: &Pool, kind: &str, repo_id: i64, target_id: i64, number: i64, title: &str, body: &str) -> sqlx::Result<()> { |
| 329 | 76× | sqlx::query("DELETE FROM search WHERE kind = $1 AND target_id = $2").bind(kind).bind(target_id).execute(pool).await?; |
| 330 | 76× | sqlx::query("INSERT INTO search (kind, repo_id, target_id, number, title, body) VALUES ($1, $2, $3, $4, $5, $6)") |
| 331 | 76× | .bind(kind) |
| 332 | 76× | .bind(repo_id) |
| 333 | 76× | .bind(target_id) |
| 334 | 76× | .bind(number) |
| 335 | 76× | .bind(title) |
| 336 | 76× | .bind(body) |
| 337 | 76× | .execute(pool) |
| 338 | 76× | .await?; |
| 339 | 76× | Ok(()) |
| 340 | 76× | } |
| 341 | |
|
| 342 | |
|
| 343 | |
|
| 344 | | #[cfg(not(feature = "postgres"))] |
| 345 | 8× | pub async fn search(pool: &Pool, query: &str, repo_id: Option<i64>, limit: i64) -> sqlx::Result<Vec<Hit>> { |
| 346 | 8× | let terms: Vec<String> = query.split_whitespace().map(|t| format!("\"{}\"", t.replace('"', ""))).collect(); |
| 347 | 8× | if terms.is_empty() { |
| 348 | 0× | return Ok(vec![]); |
| 349 | 8× | } |
| 350 | 8× | let q = terms.join(" "); |
| 351 | 8× | let sql = "SELECT kind, repo_id, target_id, number, title, snippet(search, 5, '<mark>', '</mark>', '…', 12) AS snippet FROM search WHERE search MATCH $1 AND ($2 IS NULL OR repo_id = $3) ORDER BY rank LIMIT $4"; |
| 352 | 8× | sqlx::query_as(sql).bind(q).bind(repo_id).bind(repo_id).bind(limit).fetch_all(pool).await |
| 353 | 8× | } |
| 354 | |
|
| 355 | |
|
| 356 | |
|
| 357 | |
|
| 358 | | #[cfg(feature = "postgres")] |
| 359 | | pub async fn search(pool: &Pool, query: &str, repo_id: Option<i64>, limit: i64) -> sqlx::Result<Vec<Hit>> { |
| 360 | | if query.split_whitespace().next().is_none() { |
| 361 | | return Ok(vec![]); |
| 362 | | } |
| 363 | | let sql = "SELECT kind, repo_id, target_id, number, title, \ |
| 364 | | ts_headline('english', body, q, 'StartSel=\"<mark>\", StopSel=\"</mark>\", MaxWords=12, MinWords=4') AS snippet \ |
| 365 | | FROM search, plainto_tsquery('english', $1) q WHERE doc @@ q AND ($2 IS NULL OR repo_id = $3) \ |
| 366 | | ORDER BY ts_rank(doc, q) DESC LIMIT $4"; |
| 367 | | sqlx::query_as(sql).bind(query).bind(repo_id).bind(repo_id).bind(limit).fetch_all(pool).await |
| 368 | | } |