fabrica

hanna/fabrica

main
Switch branch
mainView all branches
Clone
HTTPS
SSH
18963 bytes
Raw
1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! Repositories: creation and path lookup.
6
7use model::{Repo, Visibility};
8use sqlx::any::AnyRow;
9use sqlx::{AssertSqlSafe, Row};
10
11use crate::{Store, StoreError, map_conflict, new_id, now_ms};
12
13/// The columns of `repos` in a fixed order, shared by every `SELECT`.
14pub(crate) const REPO_COLUMNS: &str = "id, owner_id, group_id, name, path, description, visibility, \
15 default_branch, object_format, issues_enabled, pulls_enabled, releases_enabled, is_mirror, \
16 fork_parent_id, next_iid, archived_at, size_bytes, pushed_at, created_at, updated_at";
17
18/// Fields needed to create a repository. The server fills in the id, timestamps,
19/// and the derived size/push bookkeeping.
20#[derive(Debug, Clone)]
21pub struct NewRepo {
22 /// Owning user id.
23 pub owner_id: String,
24 /// Containing group id, or `None` when the repo hangs directly off the owner.
25 pub group_id: Option<String>,
26 /// The repo's own name; validated with [`model::validate_name`].
27 pub name: String,
28 /// Materialized full path within the owner (`group/sub/repo`, or just `name`).
29 pub path: String,
30 /// Optional one-line description.
31 pub description: Option<String>,
32 /// The repo's visibility (public / internal / private).
33 pub visibility: Visibility,
34 /// The repo's default branch.
35 pub default_branch: String,
36}
37
38/// A repo collaborator with their username resolved, for listings.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct Collaborator {
41 /// The collaborator's user id.
42 pub user_id: String,
43 /// Their username.
44 pub username: String,
45 /// Their permission (`read` | `write` | `admin`).
46 pub permission: String,
47}
48
49impl Store {
50 /// Create a repository, returning the persisted row.
51 ///
52 /// # Errors
53 ///
54 /// * [`StoreError::Name`] if the repo name is invalid.
55 /// * [`StoreError::Conflict`] if the owner already has a repo at that path.
56 /// * [`StoreError::Query`] for any other database failure.
57 pub async fn create_repo(&self, new: NewRepo) -> Result<Repo, StoreError> {
58 model::validate_name(&new.name)?;
59 let now = now_ms();
60 let repo = Repo {
61 id: new_id(),
62 owner_id: new.owner_id,
63 group_id: new.group_id,
64 name: new.name,
65 path: new.path,
66 description: new.description,
67 visibility: new.visibility,
68 default_branch: new.default_branch,
69 object_format: "sha1".to_string(),
70 issues_enabled: true,
71 pulls_enabled: true,
72 releases_enabled: true,
73 is_mirror: false,
74 fork_parent_id: None,
75 next_iid: 1,
76 archived_at: None,
77 size_bytes: 0,
78 pushed_at: None,
79 created_at: now,
80 updated_at: now,
81 };
82
83 let sql = "INSERT INTO repos \
84 (id, owner_id, group_id, name, path, description, visibility, default_branch, \
85 archived_at, size_bytes, pushed_at, created_at, updated_at) \
86 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)";
87 sqlx::query(sql)
88 .bind(repo.id.clone())
89 .bind(repo.owner_id.clone())
90 .bind(repo.group_id.clone())
91 .bind(repo.name.clone())
92 .bind(repo.path.clone())
93 .bind(repo.description.clone())
94 .bind(repo.visibility.as_str())
95 .bind(repo.default_branch.clone())
96 .bind(repo.archived_at)
97 .bind(repo.size_bytes)
98 .bind(repo.pushed_at)
99 .bind(repo.created_at)
100 .bind(repo.updated_at)
101 .execute(&self.pool)
102 .await
103 .map_err(|e| map_conflict(e, "repo", &[("path", "path")]))?;
104
105 Ok(repo)
106 }
107
108 /// Set a repository's object format (`sha1` | `sha256`). Called right after
109 /// [`Store::create_repo`] when a non-default format was chosen, before the
110 /// on-disk repository is created, so the row matches the on-disk repo.
111 ///
112 /// # Errors
113 ///
114 /// Returns [`StoreError::Query`] on a database failure.
115 pub async fn set_object_format(
116 &self,
117 repo_id: &str,
118 object_format: &str,
119 ) -> Result<(), StoreError> {
120 sqlx::query("UPDATE repos SET object_format = $1 WHERE id = $2")
121 .bind(object_format)
122 .bind(repo_id)
123 .execute(&self.pool)
124 .await?;
125 Ok(())
126 }
127
128 /// Resolve a repository by `(owner_id, path)`, or `None` if absent. This is
129 /// the authoritative name→repo lookup; the on-disk tree is keyed by id.
130 ///
131 /// # Errors
132 ///
133 /// Returns [`StoreError::Query`] on a database failure.
134 pub async fn repo_by_owner_path(
135 &self,
136 owner_id: &str,
137 path: &str,
138 ) -> Result<Option<Repo>, StoreError> {
139 let sql = format!("SELECT {REPO_COLUMNS} FROM repos WHERE owner_id = $1 AND path = $2");
140 let row = sqlx::query(AssertSqlSafe(sql))
141 .bind(owner_id)
142 .bind(path)
143 .fetch_optional(&self.pool)
144 .await?;
145 Ok(row.as_ref().map(|r| self.map_repo(r)).transpose()?)
146 }
147
148 /// List every repository owned by `owner_id`, ordered by path. Used by
149 /// `fabrica user del` to refuse (and enumerate) when a user still owns repos.
150 ///
151 /// # Errors
152 ///
153 /// Returns [`StoreError::Query`] on a database failure.
154 pub async fn repos_by_owner(&self, owner_id: &str) -> Result<Vec<Repo>, StoreError> {
155 let sql = format!("SELECT {REPO_COLUMNS} FROM repos WHERE owner_id = $1 ORDER BY path ASC");
156 let rows = sqlx::query(AssertSqlSafe(sql))
157 .bind(owner_id)
158 .fetch_all(&self.pool)
159 .await?;
160 rows.iter()
161 .map(|r| self.map_repo(r))
162 .collect::<Result<_, _>>()
163 .map_err(Into::into)
164 }
165
166 /// Fetch a repository by id, or `None`.
167 ///
168 /// # Errors
169 ///
170 /// Returns [`StoreError::Query`] on a database failure.
171 pub async fn repo_by_id(&self, id: &str) -> Result<Option<Repo>, StoreError> {
172 let sql = format!("SELECT {REPO_COLUMNS} FROM repos WHERE id = $1");
173 let row = sqlx::query(AssertSqlSafe(sql))
174 .bind(id)
175 .fetch_optional(&self.pool)
176 .await?;
177 Ok(row.as_ref().map(|r| self.map_repo(r)).transpose()?)
178 }
179
180 /// List every repository, ordered by owner then path. Used by `repo list`.
181 ///
182 /// # Errors
183 ///
184 /// Returns [`StoreError::Query`] on a database failure.
185 pub async fn list_repos(&self) -> Result<Vec<Repo>, StoreError> {
186 let sql = format!("SELECT {REPO_COLUMNS} FROM repos ORDER BY owner_id ASC, path ASC");
187 let rows = sqlx::query(AssertSqlSafe(sql))
188 .fetch_all(&self.pool)
189 .await?;
190 rows.iter()
191 .map(|r| self.map_repo(r))
192 .collect::<Result<_, _>>()
193 .map_err(Into::into)
194 }
195
196 /// One page of repositories, ordered by owner then path. For the admin view.
197 ///
198 /// # Errors
199 ///
200 /// Returns [`StoreError::Query`] on a database failure.
201 pub async fn list_repos_paged(&self, limit: i64, offset: i64) -> Result<Vec<Repo>, StoreError> {
202 let sql = format!(
203 "SELECT {REPO_COLUMNS} FROM repos ORDER BY owner_id ASC, path ASC LIMIT $1 OFFSET $2"
204 );
205 let rows = sqlx::query(AssertSqlSafe(sql))
206 .bind(limit)
207 .bind(offset)
208 .fetch_all(&self.pool)
209 .await?;
210 rows.iter()
211 .map(|r| self.map_repo(r))
212 .collect::<Result<_, _>>()
213 .map_err(Into::into)
214 }
215
216 /// Rename a repository: update its leaf `name` and materialized `path`
217 /// (metadata only — the on-disk directory is keyed by id and never moves).
218 ///
219 /// # Errors
220 ///
221 /// * [`StoreError::Name`] if `new_name` is invalid.
222 /// * [`StoreError::Conflict`] if the owner already has a repo at `new_path`.
223 /// * [`StoreError::Query`] on any other database failure.
224 pub async fn rename_repo(
225 &self,
226 repo_id: &str,
227 new_name: &str,
228 new_path: &str,
229 ) -> Result<bool, StoreError> {
230 model::validate_name(new_name)?;
231 let updated =
232 sqlx::query("UPDATE repos SET name = $1, path = $2, updated_at = $3 WHERE id = $4")
233 .bind(new_name)
234 .bind(new_path)
235 .bind(now_ms())
236 .bind(repo_id)
237 .execute(&self.pool)
238 .await
239 .map_err(|e| map_conflict(e, "repo", &[("path", "path")]))?
240 .rows_affected();
241 Ok(updated > 0)
242 }
243
244 /// Set a repository's visibility (private vs public). Returns `true` if the
245 /// row existed and was updated.
246 ///
247 /// # Errors
248 ///
249 /// Returns [`StoreError::Query`] on a database failure.
250 pub async fn set_visibility(
251 &self,
252 repo_id: &str,
253 visibility: Visibility,
254 ) -> Result<bool, StoreError> {
255 let updated =
256 sqlx::query("UPDATE repos SET visibility = $1, updated_at = $2 WHERE id = $3")
257 .bind(visibility.as_str())
258 .bind(now_ms())
259 .bind(repo_id)
260 .execute(&self.pool)
261 .await?
262 .rows_affected();
263 Ok(updated > 0)
264 }
265
266 /// Enable or disable issues / pull requests for a repository. `feature` is
267 /// `"issues"` or `"pulls"`. Returns `true` if the row was updated.
268 ///
269 /// # Errors
270 ///
271 /// Returns [`StoreError::Query`] on a database failure.
272 pub async fn set_feature_enabled(
273 &self,
274 repo_id: &str,
275 feature: &str,
276 enabled: bool,
277 ) -> Result<bool, StoreError> {
278 let column = match feature {
279 "issues" => "issues_enabled",
280 "pulls" => "pulls_enabled",
281 "releases" => "releases_enabled",
282 _ => return Ok(false),
283 };
284 let sql = format!("UPDATE repos SET {column} = $1, updated_at = $2 WHERE id = $3");
285 let updated = sqlx::query(AssertSqlSafe(sql))
286 .bind(enabled)
287 .bind(now_ms())
288 .bind(repo_id)
289 .execute(&self.pool)
290 .await?
291 .rows_affected();
292 Ok(updated > 0)
293 }
294
295 /// Mark a repository as a mirror (or not). Mirror repos are kept in sync from
296 /// a remote and have issues/PRs disabled.
297 ///
298 /// # Errors
299 ///
300 /// Returns [`StoreError::Query`] on a database failure.
301 pub async fn set_mirror(&self, repo_id: &str, is_mirror: bool) -> Result<bool, StoreError> {
302 let updated = sqlx::query("UPDATE repos SET is_mirror = $1, updated_at = $2 WHERE id = $3")
303 .bind(is_mirror)
304 .bind(now_ms())
305 .bind(repo_id)
306 .execute(&self.pool)
307 .await?
308 .rows_affected();
309 Ok(updated > 0)
310 }
311
312 /// Record that `repo_id` was forked from `parent_id`.
313 ///
314 /// # Errors
315 ///
316 /// Returns [`StoreError::Query`] on a database failure.
317 pub async fn set_fork_parent(&self, repo_id: &str, parent_id: &str) -> Result<(), StoreError> {
318 sqlx::query("UPDATE repos SET fork_parent_id = $1 WHERE id = $2")
319 .bind(parent_id)
320 .bind(repo_id)
321 .execute(&self.pool)
322 .await?;
323 Ok(())
324 }
325
326 /// The number of repositories forked from `repo_id`.
327 ///
328 /// # Errors
329 ///
330 /// Returns [`StoreError::Query`] on a database failure.
331 pub async fn count_forks(&self, repo_id: &str) -> Result<i64, StoreError> {
332 let n: i64 = sqlx::query("SELECT COUNT(*) AS n FROM repos WHERE fork_parent_id = $1")
333 .bind(repo_id)
334 .fetch_one(&self.pool)
335 .await?
336 .try_get("n")?;
337 Ok(n)
338 }
339
340 /// Archive or unarchive a repository. Archiving stamps `archived_at`;
341 /// unarchiving clears it. Returns `true` if the row existed and was updated.
342 ///
343 /// # Errors
344 ///
345 /// Returns [`StoreError::Query`] on a database failure.
346 pub async fn set_archived(&self, repo_id: &str, archived: bool) -> Result<bool, StoreError> {
347 let now = now_ms();
348 let archived_at = archived.then_some(now);
349 let updated =
350 sqlx::query("UPDATE repos SET archived_at = $1, updated_at = $2 WHERE id = $3")
351 .bind(archived_at)
352 .bind(now)
353 .bind(repo_id)
354 .execute(&self.pool)
355 .await?
356 .rows_affected();
357 Ok(updated > 0)
358 }
359
360 /// Record that a repository was just pushed to: stamp `pushed_at` and refresh
361 /// `size_bytes`. Called by `fabrica hook post-receive`.
362 ///
363 /// # Errors
364 ///
365 /// Returns [`StoreError::Query`] on a database failure.
366 pub async fn record_push(&self, repo_id: &str, size_bytes: i64) -> Result<bool, StoreError> {
367 let now = now_ms();
368 let updated = sqlx::query(
369 "UPDATE repos SET pushed_at = $1, size_bytes = $2, updated_at = $1 WHERE id = $3",
370 )
371 .bind(now)
372 .bind(size_bytes)
373 .bind(repo_id)
374 .execute(&self.pool)
375 .await?
376 .rows_affected();
377 Ok(updated > 0)
378 }
379
380 /// Delete a repository row (cascading to collaborators). The caller is
381 /// responsible for the on-disk directory — moving it to trash, or purging it.
382 /// Returns `true` if a row was removed. Empty auto-created groups the repo
383 /// left behind are garbage-collected up the tree.
384 ///
385 /// # Errors
386 ///
387 /// Returns [`StoreError::Query`] on a database failure.
388 pub async fn delete_repo(&self, repo_id: &str) -> Result<bool, StoreError> {
389 // Remember the group before the row disappears, to garbage-collect it.
390 let group_id = self.repo_by_id(repo_id).await?.and_then(|r| r.group_id);
391 let deleted = sqlx::query("DELETE FROM repos WHERE id = $1")
392 .bind(repo_id)
393 .execute(&self.pool)
394 .await?
395 .rows_affected();
396 if deleted > 0 {
397 self.gc_empty_groups(group_id).await?;
398 }
399 Ok(deleted > 0)
400 }
401
402 /// The viewer's collaborator permission on a repo (`read`|`write`|`admin`), or
403 /// `None` if they are not an explicit collaborator. Feeds [`auth::access`].
404 ///
405 /// # Errors
406 ///
407 /// Returns [`StoreError::Query`] on a database failure.
408 pub async fn collaborator_permission(
409 &self,
410 repo_id: &str,
411 user_id: &str,
412 ) -> Result<Option<String>, StoreError> {
413 let row = sqlx::query(
414 "SELECT permission FROM repo_collaborators WHERE repo_id = $1 AND user_id = $2",
415 )
416 .bind(repo_id)
417 .bind(user_id)
418 .fetch_optional(&self.pool)
419 .await?;
420 Ok(row.as_ref().map(|r| r.try_get("permission")).transpose()?)
421 }
422
423 /// List a repo's explicit collaborators, alphabetical by username.
424 ///
425 /// # Errors
426 ///
427 /// Returns [`StoreError::Query`] on a database failure.
428 pub async fn list_collaborators(&self, repo_id: &str) -> Result<Vec<Collaborator>, StoreError> {
429 let sql = "SELECT c.user_id, u.username, c.permission \
430 FROM repo_collaborators c JOIN users u ON u.id = c.user_id \
431 WHERE c.repo_id = $1 ORDER BY u.username_lower ASC";
432 let rows = sqlx::query(sql).bind(repo_id).fetch_all(&self.pool).await?;
433 rows.iter()
434 .map(|r| {
435 Ok(Collaborator {
436 user_id: r.try_get("user_id")?,
437 username: r.try_get("username")?,
438 permission: r.try_get("permission")?,
439 })
440 })
441 .collect::<Result<_, sqlx::Error>>()
442 .map_err(Into::into)
443 }
444
445 /// Add or update a collaborator's permission on a repo (`read`|`write`|
446 /// `admin`). The caller validates the permission token.
447 ///
448 /// # Errors
449 ///
450 /// Returns [`StoreError::Query`] on a database failure.
451 pub async fn set_collaborator(
452 &self,
453 repo_id: &str,
454 user_id: &str,
455 permission: &str,
456 ) -> Result<(), StoreError> {
457 // Portable upsert: both dialects support ON CONFLICT DO UPDATE.
458 let sql = "INSERT INTO repo_collaborators (repo_id, user_id, permission, created_at) \
459 VALUES ($1, $2, $3, $4) \
460 ON CONFLICT (repo_id, user_id) DO UPDATE SET permission = $3";
461 sqlx::query(sql)
462 .bind(repo_id)
463 .bind(user_id)
464 .bind(permission)
465 .bind(now_ms())
466 .execute(&self.pool)
467 .await?;
468 Ok(())
469 }
470
471 /// Remove a collaborator from a repo. Returns `true` if a row was removed.
472 ///
473 /// # Errors
474 ///
475 /// Returns [`StoreError::Query`] on a database failure.
476 pub async fn remove_collaborator(
477 &self,
478 repo_id: &str,
479 user_id: &str,
480 ) -> Result<bool, StoreError> {
481 let deleted =
482 sqlx::query("DELETE FROM repo_collaborators WHERE repo_id = $1 AND user_id = $2")
483 .bind(repo_id)
484 .bind(user_id)
485 .execute(&self.pool)
486 .await?
487 .rows_affected();
488 Ok(deleted > 0)
489 }
490
491 /// Map a `repos` row (selected via [`REPO_COLUMNS`]) to a [`Repo`].
492 pub(crate) fn map_repo(&self, row: &AnyRow) -> Result<Repo, sqlx::Error> {
493 Ok(Repo {
494 id: row.try_get("id")?,
495 owner_id: row.try_get("owner_id")?,
496 group_id: row.try_get("group_id")?,
497 name: row.try_get("name")?,
498 path: row.try_get("path")?,
499 description: row.try_get("description")?,
500 visibility: Visibility::from_token(&row.try_get::<String, _>("visibility")?)
501 .unwrap_or(Visibility::Private),
502 default_branch: row.try_get("default_branch")?,
503 object_format: row.try_get("object_format")?,
504 issues_enabled: self.get_bool(row, "issues_enabled")?,
505 pulls_enabled: self.get_bool(row, "pulls_enabled")?,
506 releases_enabled: self.get_bool(row, "releases_enabled")?,
507 is_mirror: self.get_bool(row, "is_mirror")?,
508 fork_parent_id: row.try_get("fork_parent_id")?,
509 next_iid: row.try_get("next_iid")?,
510 archived_at: row.try_get("archived_at")?,
511 size_bytes: row.try_get("size_bytes")?,
512 pushed_at: row.try_get("pushed_at")?,
513 created_at: row.try_get("created_at")?,
514 updated_at: row.try_get("updated_at")?,
515 })
516 }
517}