OXYGENHUB
# 2f79b17e089a oxygenhub / crates / hub / src / db / activity.rs
9.3 KiB · Rust · last changed An assistant drafts a new repository's starting files; your Flow runs over the API; 0.9.13 (closes #347, refs andymccall/oxygenhub-android#26) 8 days ago source blame history raw
1 
//! Cross-repository reads for the signed-in dashboard (hub spec §6):
2 
//! the activity feed and the proposals waiting on a reviewer. Callers
3 
//! pass the repository ids the viewer may see, so nothing here decides
4 
//! access.
5
6 
use super::Pool;
7 
use sqlx::FromRow;
8
9 
/// One line of the activity feed.
10 
#[derive(Clone, Debug, FromRow)]
11 
pub struct Activity {
12 
    pub owner: String,
13 
    pub name: String,
14 
    pub at_ms: i64,
15 
    pub actor: String,
16 
    /// `push`, `release`, `proposal` or `issue`.
17 
    pub kind: String,
18 
    /// What happened: the event name, or the bookmark for a push.
19 
    pub event: String,
20 
    /// The proposal or issue title, the release tag.
21 
    pub title: String,
22 
    /// The proposal or issue number, 0 otherwise.
23 
    pub number: i64,
24 
}
25
26 
impl Activity {
27 
    /// Where the line links to.
28 
    pub fn url(&self) -> String {
29 
        let base = format!("/{}/{}", self.owner, self.name);
30 
        match self.kind.as_str() {
31 
            "proposal" => format!("{base}/proposal/{}", self.number),
32 
            "issue" => format!("{base}/issue/{}", self.number),
33 
            "release" => format!("{base}/releases/tag/{}", self.title),
34 
            _ => format!("{base}/pushes"),
35 
        }
36 
    }
37 
}
38
39 
fn ids_json(ids: &[i64]) -> String {
40 
    serde_json::to_string(ids).unwrap_or_else(|_| "[]".into())
41 
}
42
43 
/// The SQL with each `SELECT value FROM json_each($n)` (the ids bound as a
44 
/// JSON array) in this engine's words: Postgres has `json_array_elements_text`.
45 
fn each(sql: &'static str) -> std::borrow::Cow<'static, str> {
46 
    if cfg!(feature = "postgres") {
47 
        let mut out = sql.to_string();
48 
        for n in 1..=4 {
49 
            out = out.replace(
50 
                &format!("SELECT value FROM json_each(${n})"),
51 
                &format!("SELECT CAST(value AS BIGINT) FROM json_array_elements_text(CAST(${n} AS json)) AS t(value)"),
52 
            );
53 
        }
54 
        out.into()
55 
    } else {
56 
        sql.into()
57 
    }
58 
}
59
60 
/// The newest `limit` events across the repositories in `repo_ids`.
61 
pub async fn feed(pool: &Pool, repo_ids: &[i64], limit: i64) -> sqlx::Result<Vec<Activity>> {
62 
    feed_by(pool, repo_ids, None, limit).await
63 
}
64
65 
/// Like `feed`, only what `actor` did when one is given (profile pages).
66 
pub async fn feed_by(pool: &Pool, repo_ids: &[i64], actor: Option<&str>, limit: i64) -> sqlx::Result<Vec<Activity>> {
67 
    sqlx::query_as(
68 
        &each("WITH seen(id) AS (SELECT value FROM json_each($1)), ev AS ( \
69 
           SELECT repo_id, at_ms, pusher AS actor, 'push' AS kind, reference AS event, '' AS title, 0 AS number FROM pushes WHERE ok = TRUE \
70 
           UNION ALL SELECT repo_id, created_ms, author, 'release', 'published', tag, 0 FROM releases \
71 
           UNION ALL SELECT p.repo_id, e.at_ms, e.actor, 'proposal', e.event, p.title, p.number FROM proposal_events e JOIN proposals p ON p.id = e.proposal_id \
72 
           UNION ALL SELECT i.repo_id, e.at_ms, e.actor, 'issue', e.event, i.title, i.number FROM issue_events e JOIN issues i ON i.id = e.issue_id) \
73 
         SELECT r.owner, r.name, ev.at_ms, ev.actor, ev.kind, ev.event, ev.title, ev.number \
74 
         FROM ev JOIN repositories r ON r.id = ev.repo_id WHERE ev.repo_id IN (SELECT id FROM seen) \
75 
           AND ($2 IS NULL OR lower(ev.actor) = lower($3)) \
76 
         ORDER BY ev.at_ms DESC LIMIT $4"),
77 
    )
78 
    .bind(ids_json(repo_ids))
79 
    .bind(actor)
80 
    .bind(actor)
81 
    .bind(limit)
82 
    .fetch_all(pool)
83 
    .await
84 
}
85
86 
/// A proposal on the dashboard, with its repository's name.
87 
#[derive(Clone, Debug, FromRow)]
88 
pub struct ProposalLine {
89 
    pub owner: String,
90 
    pub name: String,
91 
    pub number: i64,
92 
    pub title: String,
93 
    pub author: String,
94 
    pub state: String,
95 
    pub conflicted: bool,
96 
    pub updated_ms: i64,
97 
}
98
99 
/// Open proposals in `repo_ids` by someone else that `account_id` has
100 
/// not reviewed yet: the review queue.
101 
pub async fn awaiting_review(pool: &Pool, account_id: i64, repo_ids: &[i64]) -> sqlx::Result<Vec<ProposalLine>> {
102 
    sqlx::query_as(&each(
103 
        "SELECT r.owner, r.name, p.number, p.title, p.author, p.state, p.conflicted, p.updated_ms \
104 
         FROM proposals p JOIN repositories r ON r.id = p.repo_id \
105 
         WHERE p.repo_id IN (SELECT value FROM json_each($1)) AND p.state = 'open' \
106 
           AND (p.author_id IS NULL OR p.author_id != $2) \
107 
           AND NOT EXISTS (SELECT 1 FROM proposal_reviews v WHERE v.proposal_id = p.id AND v.author_id = $3) \
108 
         ORDER BY p.updated_ms DESC",
109 
    ))
110 
    .bind(ids_json(repo_ids))
111 
    .bind(account_id)
112 
    .bind(account_id)
113 
    .fetch_all(pool)
114 
    .await
115 
}
116
117 
/// `account_id`'s own open and draft proposals in repositories it can
118 
/// still see.
119 
pub async fn authored(pool: &Pool, account_id: i64, repo_ids: &[i64]) -> sqlx::Result<Vec<ProposalLine>> {
120 
    sqlx::query_as(&each(
121 
        "SELECT r.owner, r.name, p.number, p.title, p.author, p.state, p.conflicted, p.updated_ms \
122 
         FROM proposals p JOIN repositories r ON r.id = p.repo_id \
123 
         WHERE p.author_id = $1 AND p.state IN ('open', 'draft') AND p.repo_id IN (SELECT value FROM json_each($2)) \
124 
         ORDER BY p.updated_ms DESC",
125 
    ))
126 
    .bind(account_id)
127 
    .bind(ids_json(repo_ids))
128 
    .fetch_all(pool)
129 
    .await
130 
}
131
132 
/// An issue on the dashboard, with its repository's name.
133 
#[derive(Clone, Debug, FromRow)]
134 
pub struct IssueLine {
135 
    pub owner: String,
136 
    pub name: String,
137 
    pub number: i64,
138 
    pub title: String,
139 
    pub updated_ms: i64,
140 
}
141
142 
/// Open issues assigned to `account_id`, in repositories it can see.
143 
pub async fn assigned(pool: &Pool, account_id: i64, repo_ids: &[i64]) -> sqlx::Result<Vec<IssueLine>> {
144 
    sqlx::query_as(&each(
145 
        "SELECT r.owner, r.name, i.number, i.title, i.updated_ms \
146 
         FROM issues i JOIN issue_assignees a ON a.issue_id = i.id JOIN repositories r ON r.id = i.repo_id \
147 
         WHERE a.account_id = $1 AND i.state = 'open' AND i.repo_id IN (SELECT value FROM json_each($2)) \
148 
         ORDER BY i.updated_ms DESC",
149 
    ))
150 
    .bind(account_id)
151 
    .bind(ids_json(repo_ids))
152 
    .fetch_all(pool)
153 
    .await
154 
}
155
156 
/// One event of Your sky (docs/backgrounds.md §5): its repository, when,
157 
/// its kind (0 a push, 1 a proposal event, 2 an issue event, 3 a release,
158 
/// 4 a Flow run) and how many marks it makes: a push's changes, else one.
159 
#[derive(Clone, Debug, FromRow)]
160 
pub struct SkyEvent {
161 
    pub repo_id: i64,
162 
    pub at_ms: i64,
163 
    pub kind: i64,
164 
    pub n: i64,
165 
}
166
167 
/// The events since `since_ms` in `repo_ids`, newest first, at most `limit`.
168 
pub async fn sky(pool: &Pool, repo_ids: &[i64], since_ms: i64, limit: i64) -> sqlx::Result<Vec<SkyEvent>> {
169 
    sqlx::query_as(
170 
        &each("WITH seen(id) AS (SELECT value FROM json_each($1)), ev AS ( \
171 
           SELECT repo_id, at_ms, CAST(0 AS BIGINT) AS kind, CASE WHEN changes > 1 THEN changes ELSE 1 END AS n FROM pushes WHERE ok = TRUE \
172 
           UNION ALL SELECT p.repo_id, e.at_ms, 1, 1 FROM proposal_events e JOIN proposals p ON p.id = e.proposal_id \
173 
           UNION ALL SELECT i.repo_id, e.at_ms, 2, 1 FROM issue_events e JOIN issues i ON i.id = e.issue_id \
174 
           UNION ALL SELECT repo_id, created_ms, 3, 1 FROM releases \
175 
           UNION ALL SELECT repo_id, created_ms, 4, 1 FROM flow_runs) \
176 
         SELECT repo_id, at_ms, kind, n FROM ev WHERE repo_id IN (SELECT id FROM seen) AND at_ms >= $2 \
177 
         ORDER BY at_ms DESC LIMIT $3"),
178 
    )
179 
    .bind(ids_json(repo_ids))
180 
    .bind(since_ms)
181 
    .bind(limit)
182 
    .fetch_all(pool)
183 
    .await
184 
}
185
186 
/// One account's pushes of the last year in the repositories a viewer may
187 
/// see, for the profile's pulse (#217): `(repo_id, at_ms, changes, via)`.
188 
/// Landings are left out (their changes were pushed once already), as are
189 
/// failed pushes and pushes that brought nothing.
190 
pub async fn pulse(pool: &Pool, account_id: i64, repo_ids: &[i64], since_ms: i64) -> sqlx::Result<Vec<(i64, i64, i64, String)>> {
191 
    let ids = serde_json::to_string(repo_ids).unwrap_or_else(|_| "[]".into());
192 
    sqlx::query_as(&each(
193 
        "SELECT repo_id, at_ms, changes, via FROM pushes \
194 
         WHERE account_id = $1 AND ok = TRUE AND changes > 0 AND at_ms >= $2 AND reason NOT LIKE 'landed%' \
195 
           AND repo_id IN (SELECT value FROM json_each($3)) ORDER BY at_ms",
196 
    ))
197 
    .bind(account_id)
198 
    .bind(since_ms)
199 
    .bind(ids)
200 
    .fetch_all(pool)
201 
    .await
202 
}
203
204 
/// Releases an account made in the last year in repositories a viewer may
205 
/// see: `(repo_id, tag, created_ms)`, to label the pulse's peaks.
206 
pub async fn releases_by(pool: &Pool, account_id: i64, repo_ids: &[i64], since_ms: i64) -> sqlx::Result<Vec<(i64, String, i64)>> {
207 
    let ids = serde_json::to_string(repo_ids).unwrap_or_else(|_| "[]".into());
208 
    sqlx::query_as(&each(
209 
        "SELECT repo_id, tag, created_ms FROM releases WHERE author_id = $1 AND created_ms >= $2 \
210 
           AND repo_id IN (SELECT value FROM json_each($3)) ORDER BY created_ms",
211 
    ))
212 
    .bind(account_id)
213 
    .bind(since_ms)
214 
    .bind(ids)
215 
    .fetch_all(pool)
216 
    .await
217 
}
218
219 
/// The newest Flow runs of `repo_ids`, created before `before_ms`, at most
220 
/// `limit` (oxygenhub-android#26: the app's Flow monitor).
221 
pub async fn recent_runs(pool: &Pool, repo_ids: &[i64], before_ms: i64, limit: i64) -> sqlx::Result<Vec<super::flow::Run>> {
222 
    sqlx::query_as(&each(
223 
        "WITH seen(id) AS (SELECT value FROM json_each($1)) \
224 
         SELECT * FROM flow_runs WHERE repo_id IN (SELECT id FROM seen) AND created_ms < $2 \
225 
         ORDER BY created_ms DESC, id DESC LIMIT $3",
226 
    ))
227 
    .bind(ids_json(repo_ids))
228 
    .bind(before_ms)
229 
    .bind(limit)
230 
    .fetch_all(pool)
231 
    .await
232 
}