OXYGENHUB
# 3d3d29fbcb00 oxygenhub / crates / hub / src / db / proposals.rs
9.4 KiB · Rust · last changed Write transactions begin IMMEDIATE, so a concurrent writer is waited for (busy timeout) instead of failing at once with database is locked; a test of 16 concurrent log appends fails without it 1 month ago source blame history raw
1 
//! The `proposals`, `proposal_comments`, `proposal_reviews` and
2 
//! `proposal_events` tables, and the per-repository number space.
3
4 
use super::{now_ms, Pool};
5 
use sqlx::FromRow;
6
7 
#[derive(Clone, Debug, FromRow)]
8 
pub struct ProposalRow {
9 
    pub id: i64,
10 
    pub repo_id: i64,
11 
    pub number: i64,
12 
    pub title: String,
13 
    pub body: String,
14 
    pub author_id: Option<i64>,
15 
    pub author: String,
16 
    pub state: String,
17 
    pub target: String,
18 
    pub source_kind: String,
19 
    pub source_repo_id: Option<i64>,
20 
    pub source_bookmark: String,
21 
    pub source_changes: String,
22 
    pub projected_state: String,
23 
    pub projected_heads: String,
24 
    pub projected_target: String,
25 
    pub projected_changes: String,
26 
    pub conflicted: bool,
27 
    pub landed_state: String,
28 
    pub created_ms: i64,
29 
    pub updated_ms: i64,
30 
}
31
32 
impl ProposalRow {
33 
    pub fn projected_heads(&self) -> Vec<String> {
34 
        self.projected_heads.split(',').filter(|s| !s.is_empty()).map(String::from).collect()
35 
    }
36
37 
    pub fn projected_changes(&self) -> Vec<String> {
38 
        self.projected_changes.split(',').filter(|s| !s.is_empty()).map(String::from).collect()
39 
    }
40
41 
    pub fn source_changes(&self) -> Vec<String> {
42 
        self.source_changes.split(',').filter(|s| !s.is_empty()).map(String::from).collect()
43 
    }
44
45 
    pub fn is_open(&self) -> bool {
46 
        self.state == "open" || self.state == "draft"
47 
    }
48 
}
49
50 
#[derive(Clone, Debug, FromRow)]
51 
pub struct CommentRow {
52 
    pub id: i64,
53 
    pub proposal_id: i64,
54 
    pub author_id: Option<i64>,
55 
    pub author: String,
56 
    pub body: String,
57 
    pub path: String,
58 
    pub anchor_change: String,
59 
    pub anchor_index: i64,
60 
    pub resolved: bool,
61 
    pub created_ms: i64,
62 
}
63
64 
#[derive(Clone, Debug, FromRow)]
65 
pub struct ReviewRow {
66 
    pub id: i64,
67 
    pub proposal_id: i64,
68 
    pub author_id: i64,
69 
    pub author: String,
70 
    pub verdict: String,
71 
    pub body: String,
72 
    pub created_ms: i64,
73 
}
74
75 
#[derive(Clone, Debug, FromRow)]
76 
pub struct EventRow {
77 
    pub id: i64,
78 
    pub at_ms: i64,
79 
    pub actor: String,
80 
    pub event: String,
81 
    pub detail: String,
82 
}
83
84 
/// Take the next number of a repository's shared number space.
85 
pub async fn next_number(pool: &Pool, repo_id: i64) -> sqlx::Result<i64> {
86 
    let mut tx = super::begin_write(pool).await?;
87 
    sqlx::query("INSERT OR IGNORE INTO numbers (repo_id, next) VALUES (?, 1)").bind(repo_id).execute(&mut *tx).await?;
88 
    let n: i64 = sqlx::query_scalar("SELECT next FROM numbers WHERE repo_id = ?").bind(repo_id).fetch_one(&mut *tx).await?;
89 
    sqlx::query("UPDATE numbers SET next = next + 1 WHERE repo_id = ?").bind(repo_id).execute(&mut *tx).await?;
90 
    tx.commit().await?;
91 
    Ok(n)
92 
}
93
94 
pub struct NewProposal<'a> {
95 
    pub repo_id: i64,
96 
    pub title: &'a str,
97 
    pub body: &'a str,
98 
    pub author_id: Option<i64>,
99 
    pub author: &'a str,
100 
    pub draft: bool,
101 
    pub target: &'a str,
102 
    pub source_kind: &'a str,
103 
    pub source_repo_id: Option<i64>,
104 
    pub source_bookmark: &'a str,
105 
    pub source_changes: &'a str,
106 
}
107
108 
pub async fn create(pool: &Pool, p: NewProposal<'_>) -> sqlx::Result<ProposalRow> {
109 
    let number = next_number(pool, p.repo_id).await?;
110 
    let now = now_ms();
111 
    sqlx::query_as(
112 
        "INSERT INTO proposals (repo_id, number, title, body, author_id, author, state, target, source_kind, source_repo_id, source_bookmark, source_changes, created_ms, updated_ms) \
113 
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING *",
114 
    )
115 
    .bind(p.repo_id)
116 
    .bind(number)
117 
    .bind(p.title)
118 
    .bind(p.body)
119 
    .bind(p.author_id)
120 
    .bind(p.author)
121 
    .bind(if p.draft { "draft" } else { "open" })
122 
    .bind(p.target)
123 
    .bind(p.source_kind)
124 
    .bind(p.source_repo_id)
125 
    .bind(p.source_bookmark)
126 
    .bind(p.source_changes)
127 
    .bind(now)
128 
    .bind(now)
129 
    .fetch_one(pool)
130 
    .await
131 
}
132
133 
pub async fn get(pool: &Pool, repo_id: i64, number: i64) -> sqlx::Result<Option<ProposalRow>> {
134 
    sqlx::query_as("SELECT * FROM proposals WHERE repo_id = ? AND number = ?").bind(repo_id).bind(number).fetch_optional(pool).await
135 
}
136
137 
pub async fn list(pool: &Pool, repo_id: i64, states: &[&str]) -> sqlx::Result<Vec<ProposalRow>> {
138 
    let all: Vec<ProposalRow> =
139 
        sqlx::query_as("SELECT * FROM proposals WHERE repo_id = ? ORDER BY number DESC").bind(repo_id).fetch_all(pool).await?;
140 
    Ok(all.into_iter().filter(|p| states.is_empty() || states.contains(&p.state.as_str())).collect())
141 
}
142
143 
/// Open proposals of every repository, for the front page.
144 
pub async fn open_for_author(pool: &Pool, author_id: i64) -> sqlx::Result<Vec<ProposalRow>> {
145 
    sqlx::query_as("SELECT * FROM proposals WHERE author_id = ? AND state IN ('open', 'draft') ORDER BY updated_ms DESC")
146 
        .bind(author_id)
147 
        .fetch_all(pool)
148 
        .await
149 
}
150
151 
pub async fn set_projection(
152 
    pool: &Pool,
153 
    id: i64,
154 
    state: &str,
155 
    heads: &str,
156 
    target_state: &str,
157 
    changes: &str,
158 
    conflicted: bool,
159 
) -> sqlx::Result<()> {
160 
    sqlx::query("UPDATE proposals SET projected_state = ?, projected_heads = ?, projected_target = ?, projected_changes = ?, conflicted = ? WHERE id = ?")
161 
        .bind(state)
162 
        .bind(heads)
163 
        .bind(target_state)
164 
        .bind(changes)
165 
        .bind(conflicted)
166 
        .bind(id)
167 
        .execute(pool)
168 
        .await?;
169 
    Ok(())
170 
}
171
172 
pub async fn set_state(pool: &Pool, id: i64, state: &str, landed_state: &str) -> sqlx::Result<()> {
173 
    sqlx::query("UPDATE proposals SET state = ?, landed_state = ?, updated_ms = ? WHERE id = ?")
174 
        .bind(state)
175 
        .bind(landed_state)
176 
        .bind(now_ms())
177 
        .bind(id)
178 
        .execute(pool)
179 
        .await?;
180 
    Ok(())
181 
}
182
183 
pub async fn edit(pool: &Pool, id: i64, title: &str, body: &str, target: &str) -> sqlx::Result<()> {
184 
    sqlx::query("UPDATE proposals SET title = ?, body = ?, target = ?, updated_ms = ? WHERE id = ?")
185 
        .bind(title)
186 
        .bind(body)
187 
        .bind(target)
188 
        .bind(now_ms())
189 
        .bind(id)
190 
        .execute(pool)
191 
        .await?;
192 
    Ok(())
193 
}
194
195 
pub async fn touch(pool: &Pool, id: i64) -> sqlx::Result<()> {
196 
    sqlx::query("UPDATE proposals SET updated_ms = ? WHERE id = ?").bind(now_ms()).bind(id).execute(pool).await?;
197 
    Ok(())
198 
}
199
200 
pub async fn comments(pool: &Pool, proposal_id: i64) -> sqlx::Result<Vec<CommentRow>> {
201 
    sqlx::query_as("SELECT * FROM proposal_comments WHERE proposal_id = ? ORDER BY created_ms, id").bind(proposal_id).fetch_all(pool).await
202 
}
203
204 
pub async fn add_comment(
205 
    pool: &Pool,
206 
    proposal_id: i64,
207 
    author_id: Option<i64>,
208 
    author: &str,
209 
    body: &str,
210 
    path: &str,
211 
    anchor: Option<(&str, i64)>,
212 
) -> sqlx::Result<CommentRow> {
213 
    let row = sqlx::query_as(
214 
        "INSERT INTO proposal_comments (proposal_id, author_id, author, body, path, anchor_change, anchor_index, created_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?) RETURNING *",
215 
    )
216 
    .bind(proposal_id)
217 
    .bind(author_id)
218 
    .bind(author)
219 
    .bind(body)
220 
    .bind(path)
221 
    .bind(anchor.map(|a| a.0).unwrap_or(""))
222 
    .bind(anchor.map(|a| a.1).unwrap_or(0))
223 
    .bind(now_ms())
224 
    .fetch_one(pool)
225 
    .await?;
226 
    touch(pool, proposal_id).await?;
227 
    Ok(row)
228 
}
229
230 
pub async fn set_comment_resolved(pool: &Pool, id: i64, resolved: bool) -> sqlx::Result<()> {
231 
    sqlx::query("UPDATE proposal_comments SET resolved = ? WHERE id = ?").bind(resolved).bind(id).execute(pool).await?;
232 
    Ok(())
233 
}
234
235 
pub async fn reviews(pool: &Pool, proposal_id: i64) -> sqlx::Result<Vec<ReviewRow>> {
236 
    sqlx::query_as("SELECT * FROM proposal_reviews WHERE proposal_id = ? ORDER BY created_ms, id").bind(proposal_id).fetch_all(pool).await
237 
}
238
239 
pub async fn add_review(pool: &Pool, proposal_id: i64, author_id: i64, author: &str, verdict: &str, body: &str) -> sqlx::Result<ReviewRow> {
240 
    let row = sqlx::query_as(
241 
        "INSERT INTO proposal_reviews (proposal_id, author_id, author, verdict, body, created_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING *",
242 
    )
243 
    .bind(proposal_id)
244 
    .bind(author_id)
245 
    .bind(author)
246 
    .bind(verdict)
247 
    .bind(body)
248 
    .bind(now_ms())
249 
    .fetch_one(pool)
250 
    .await?;
251 
    touch(pool, proposal_id).await?;
252 
    Ok(row)
253 
}
254
255 
pub async fn events(pool: &Pool, proposal_id: i64) -> sqlx::Result<Vec<EventRow>> {
256 
    sqlx::query_as("SELECT id, at_ms, actor, event, detail FROM proposal_events WHERE proposal_id = ? ORDER BY at_ms, id")
257 
        .bind(proposal_id)
258 
        .fetch_all(pool)
259 
        .await
260 
}
261
262 
pub async fn add_event(pool: &Pool, proposal_id: i64, actor: &str, event: &str, detail: &str) -> sqlx::Result<()> {
263 
    sqlx::query("INSERT INTO proposal_events (proposal_id, at_ms, actor, event, detail) VALUES (?, ?, ?, ?, ?)")
264 
        .bind(proposal_id)
265 
        .bind(now_ms())
266 
        .bind(actor)
267 
        .bind(event)
268 
        .bind(detail)
269 
        .execute(pool)
270 
        .await?;
271 
    Ok(())
272 
}
273
274 
/// Landing requirements of a repository.
275 
#[derive(Clone, Copy, Debug, Default, FromRow)]
276 
pub struct Requirements {
277 
    pub required_approvals: i64,
278 
    pub require_owners: bool,
279 
    pub allow_conflicts: bool,
280 
}
281
282 
pub async fn requirements(pool: &Pool, repo_id: i64) -> sqlx::Result<Requirements> {
283 
    sqlx::query_as("SELECT required_approvals, require_owners, allow_conflicts FROM repositories WHERE id = ?")
284 
        .bind(repo_id)
285 
        .fetch_one(pool)
286 
        .await
287 
}
288
289 
pub async fn set_requirements(pool: &Pool, repo_id: i64, r: Requirements) -> sqlx::Result<()> {
290 
    sqlx::query("UPDATE repositories SET required_approvals = ?, require_owners = ?, allow_conflicts = ? WHERE id = ?")
291 
        .bind(r.required_approvals)
292 
        .bind(r.require_owners)
293 
        .bind(r.allow_conflicts)
294 
        .bind(repo_id)
295 
        .execute(pool)
296 
        .await?;
297 
    Ok(())
298 
}