OXYGENHUB
# 44a3d2ebaddd oxygenhub / crates / hub / src / db / notify.rs
10.5 KiB · Rust · last changed Notifications by thread: participating versus watching, filters by kind and repository, read, unread and done per thread, and opening a row marks it read 1 month ago source blame history raw
1
//! Notifications, watching, subscriptions and the search index (hub spec
2
//! §10, §11).
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
/// Why someone is told: they take part in the thread, or only watch the repository.
24
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25
pub enum Reason {
26
    Participating,
27
    Watching,
28
}
29
30
impl Reason {
31
    pub fn as_str(self) -> &'static str {
32
        match self {
33
            Reason::Participating => "participating",
34
            Reason::Watching => "watching",
35
        }
36
    }
37
}
38
39
/// The thread a notification belongs to: its page, without a fragment.
40
pub fn thread_of(url: &str) -> &str {
41
    url.split('#').next().unwrap_or(url)
42
}
43
44
/// A notification to record.
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
pub async fn add(pool: &Pool, n: New<'_>) -> sqlx::Result<()> {
56
    let New { account_id, repo_id, reason, kind, title, url, actor } = n;
57
    // A new event brings a thread that was set aside back to the inbox.
58
    sqlx::query("UPDATE notifications SET done = 0 WHERE account_id = ? AND thread = ? AND done = 1")
59
        .bind(account_id)
60
        .bind(thread_of(url))
61
        .execute(pool)
62
        .await?;
63
    sqlx::query("INSERT INTO notifications (account_id, at_ms, kind, title, url, actor, repo_id, thread, reason) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")
64
        .bind(account_id)
65
        .bind(now_ms())
66
        .bind(kind)
67
        .bind(title)
68
        .bind(url)
69
        .bind(actor)
70
        .bind(repo_id)
71
        .bind(thread_of(url))
72
        .bind(reason.as_str())
73
        .execute(pool)
74
        .await?;
75
    Ok(())
76
}
77
78
/// One inbox row: a thread, shown by its newest notification.
79
#[derive(Clone, Debug, FromRow)]
80
pub struct Thread {
81
    /// The newest notification's id.
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
/// What the inbox shows.
98
#[derive(Clone, Debug, Default)]
99
pub struct Filter {
100
    pub done: bool,
101
    pub unread_only: bool,
102
    /// `participating` or `watching`.
103
    pub reason: Option<String>,
104
    /// A kind or its family (`issue` matches `issue.closed`).
105
    pub kind: Option<String>,
106
    pub repo_id: Option<i64>,
107
}
108
109
pub async fn threads(pool: &Pool, account_id: i64, f: &Filter, limit: i64) -> sqlx::Result<Vec<Thread>> {
110
    // SQLite takes bare columns from the row holding MAX(at_ms): the newest.
111
    let mut sql = String::from(
112
        "SELECT n.id, MAX(n.at_ms) AS at_ms, n.kind, n.title, n.url, n.actor, n.thread, n.repo_id, \
113
         (SELECT r.owner || '/' || r.name FROM repositories r WHERE r.id = n.repo_id) AS repo, \
114
         COUNT(*) AS count, SUM(n.read = 0) AS unread, SUM(n.reason = 'participating') > 0 AS participating, n.done \
115
         FROM notifications n WHERE n.account_id = ? AND n.done = ?",
116
    );
117
    if f.kind.is_some() {
118
        sql.push_str(" AND (n.kind = ? OR n.kind LIKE ? || '.%')");
119
    }
120
    if f.repo_id.is_some() {
121
        sql.push_str(" AND n.repo_id = ?");
122
    }
123
    sql.push_str(" GROUP BY n.thread HAVING 1");
124
    if f.unread_only {
125
        sql.push_str(" AND SUM(n.read = 0) > 0");
126
    }
127
    match f.reason.as_deref() {
128
        Some("participating") => sql.push_str(" AND SUM(n.reason = 'participating') > 0"),
129
        Some("watching") => sql.push_str(" AND SUM(n.reason = 'participating') = 0"),
130
        _ => {}
131
    }
132
    sql.push_str(" ORDER BY at_ms DESC LIMIT ?");
133
    let mut q = sqlx::query_as::<_, Thread>(&sql).bind(account_id).bind(f.done);
134
    if let Some(k) = &f.kind {
135
        q = q.bind(k).bind(k);
136
    }
137
    if let Some(r) = f.repo_id {
138
        q = q.bind(r);
139
    }
140
    q.bind(limit).fetch_all(pool).await
141
}
142
143
/// Mark a thread read or unread, done or back in the inbox.
144
pub async fn set_thread(pool: &Pool, account_id: i64, thread: &str, read: Option<bool>, done: Option<bool>) -> sqlx::Result<()> {
145
    if let Some(r) = read {
146
        if r {
147
            sqlx::query("UPDATE notifications SET read = 1 WHERE account_id = ? AND thread = ?")
148
                .bind(account_id)
149
                .bind(thread)
150
                .execute(pool)
151
                .await?;
152
        } else {
153
            // Unread: the newest notification of the thread only.
154
            sqlx::query("UPDATE notifications SET read = 0 WHERE id = (SELECT id FROM notifications WHERE account_id = ? AND thread = ? ORDER BY at_ms DESC, id DESC LIMIT 1)")
155
                .bind(account_id)
156
                .bind(thread)
157
                .execute(pool)
158
                .await?;
159
        }
160
    }
161
    if let Some(d) = done {
162
        let read_too = if d { ", read = 1" } else { "" };
163
        sqlx::query(&format!("UPDATE notifications SET done = ?{read_too} WHERE account_id = ? AND thread = ?"))
164
            .bind(d)
165
            .bind(account_id)
166
            .bind(thread)
167
            .execute(pool)
168
            .await?;
169
    }
170
    Ok(())
171
}
172
173
/// The notification with this id, if it is the account's.
174
pub async fn get(pool: &Pool, account_id: i64, id: i64) -> sqlx::Result<Option<Notification>> {
175
    sqlx::query_as("SELECT * FROM notifications WHERE account_id = ? AND id = ?").bind(account_id).bind(id).fetch_optional(pool).await
176
}
177
178
pub async fn inbox(pool: &Pool, account_id: i64, limit: i64) -> sqlx::Result<Vec<Notification>> {
179
    sqlx::query_as("SELECT * FROM notifications WHERE account_id = ? ORDER BY read, at_ms DESC, id DESC LIMIT ?")
180
        .bind(account_id)
181
        .bind(limit)
182
        .fetch_all(pool)
183
        .await
184
}
185
186
pub async fn unread(pool: &Pool, account_id: i64) -> sqlx::Result<i64> {
187
    sqlx::query_scalar("SELECT COUNT(*) FROM notifications WHERE account_id = ? AND read = 0 AND done = 0")
188
        .bind(account_id)
189
        .fetch_one(pool)
190
        .await
191
}
192
193
pub async fn mark_read(pool: &Pool, account_id: i64, id: Option<i64>) -> sqlx::Result<()> {
194
    match id {
195
        Some(id) => {
196
            sqlx::query("UPDATE notifications SET read = 1 WHERE account_id = ? AND id = ?").bind(account_id).bind(id).execute(pool).await?
197
        }
198
        None => sqlx::query("UPDATE notifications SET read = 1 WHERE account_id = ?").bind(account_id).execute(pool).await?,
199
    };
200
    Ok(())
201
}
202
203
// ---------------------------------------------------------------- watching
204
205
pub async fn watching(pool: &Pool, account_id: i64, repo_id: i64) -> sqlx::Result<bool> {
206
    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM watches WHERE account_id = ? AND repo_id = ?")
207
        .bind(account_id)
208
        .bind(repo_id)
209
        .fetch_one(pool)
210
        .await?;
211
    Ok(n > 0)
212
}
213
214
pub async fn set_watch(pool: &Pool, account_id: i64, repo_id: i64, on: bool) -> sqlx::Result<()> {
215
    if on {
216
        sqlx::query("INSERT OR IGNORE INTO watches (account_id, repo_id) VALUES (?, ?)")
217
            .bind(account_id)
218
            .bind(repo_id)
219
            .execute(pool)
220
            .await?;
221
    } else {
222
        sqlx::query("DELETE FROM watches WHERE account_id = ? AND repo_id = ?").bind(account_id).bind(repo_id).execute(pool).await?;
223
    }
224
    Ok(())
225
}
226
227
pub async fn watchers(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<i64>> {
228
    sqlx::query_scalar("SELECT account_id FROM watches WHERE repo_id = ?").bind(repo_id).fetch_all(pool).await
229
}
230
231
pub async fn subscribe(pool: &Pool, account_id: i64, kind: &str, target_id: i64) -> sqlx::Result<()> {
232
    sqlx::query("INSERT OR IGNORE INTO subscriptions (account_id, kind, target_id) VALUES (?, ?, ?)")
233
        .bind(account_id)
234
        .bind(kind)
235
        .bind(target_id)
236
        .execute(pool)
237
        .await?;
238
    Ok(())
239
}
240
241
pub async fn unsubscribe(pool: &Pool, account_id: i64, kind: &str, target_id: i64) -> sqlx::Result<()> {
242
    sqlx::query("DELETE FROM subscriptions WHERE account_id = ? AND kind = ? AND target_id = ?")
243
        .bind(account_id)
244
        .bind(kind)
245
        .bind(target_id)
246
        .execute(pool)
247
        .await?;
248
    Ok(())
249
}
250
251
pub async fn subscribed(pool: &Pool, account_id: i64, kind: &str, target_id: i64) -> sqlx::Result<bool> {
252
    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM subscriptions WHERE account_id = ? AND kind = ? AND target_id = ?")
253
        .bind(account_id)
254
        .bind(kind)
255
        .bind(target_id)
256
        .fetch_one(pool)
257
        .await?;
258
    Ok(n > 0)
259
}
260
261
pub async fn subscribers(pool: &Pool, kind: &str, target_id: i64) -> sqlx::Result<Vec<i64>> {
262
    sqlx::query_scalar("SELECT account_id FROM subscriptions WHERE kind = ? AND target_id = ?")
263
        .bind(kind)
264
        .bind(target_id)
265
        .fetch_all(pool)
266
        .await
267
}
268
269
// ------------------------------------------------------------------ search
270
271
#[derive(Clone, Debug, FromRow)]
272
pub struct Hit {
273
    pub kind: String,
274
    pub repo_id: i64,
275
    pub target_id: i64,
276
    pub number: i64,
277
    pub title: String,
278
    pub snippet: String,
279
}
280
281
/// Replace the indexed text of one thread (title plus body plus comments).
282
pub async fn index(pool: &Pool, kind: &str, repo_id: i64, target_id: i64, number: i64, title: &str, body: &str) -> sqlx::Result<()> {
283
    sqlx::query("DELETE FROM search WHERE kind = ? AND target_id = ?").bind(kind).bind(target_id).execute(pool).await?;
284
    sqlx::query("INSERT INTO search (kind, repo_id, target_id, number, title, body) VALUES (?, ?, ?, ?, ?, ?)")
285
        .bind(kind)
286
        .bind(repo_id)
287
        .bind(target_id)
288
        .bind(number)
289
        .bind(title)
290
        .bind(body)
291
        .execute(pool)
292
        .await?;
293
    Ok(())
294
}
295
296
/// Full-text hits, best first. The query is quoted so punctuation cannot
297
/// reach the FTS5 parser.
298
pub async fn search(pool: &Pool, query: &str, repo_id: Option<i64>, limit: i64) -> sqlx::Result<Vec<Hit>> {
299
    let terms: Vec<String> = query.split_whitespace().map(|t| format!("\"{}\"", t.replace('"', ""))).collect();
300
    if terms.is_empty() {
301
        return Ok(vec![]);
302
    }
303
    let q = terms.join(" ");
304
    let sql = "SELECT kind, repo_id, target_id, number, title, snippet(search, 5, '<mark>', '</mark>', '…', 12) AS snippet FROM search WHERE search MATCH ? AND (? IS NULL OR repo_id = ?) ORDER BY rank LIMIT ?";
305
    sqlx::query_as(sql).bind(q).bind(repo_id).bind(repo_id).bind(limit).fetch_all(pool).await
306
}