fabrica

hanna/fabrica

main
Switch branch
mainView all branches
Clone
HTTPS
SSH
9164 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//! Repository mirrors: outbound (push to a remote) and inbound (pull from a
6//! remote). The periodic sync loop reads [`Store::due_mirrors`]; a received push
7//! triggers [`Store::push_mirrors_on_push`]. Credentials are stored reversibly
8//! (a mirror must authenticate to its remote) — prefer a scoped token.
9
10use sqlx::Row;
11use sqlx::any::AnyRow;
12
13use crate::{Store, StoreError, new_id, now_ms};
14
15/// Whether a mirror pushes to, or pulls from, its remote.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum MirrorDirection {
18 /// Push the repo's refs out to the remote.
19 Push,
20 /// Pull the remote's refs into the repo.
21 Pull,
22}
23
24impl MirrorDirection {
25 /// The stored token.
26 #[must_use]
27 pub fn as_str(self) -> &'static str {
28 match self {
29 Self::Push => "push",
30 Self::Pull => "pull",
31 }
32 }
33
34 /// Parse a stored token.
35 #[must_use]
36 pub fn parse(s: &str) -> Self {
37 if s == "pull" { Self::Pull } else { Self::Push }
38 }
39}
40
41/// A configured mirror.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct Mirror {
44 /// ULID primary key.
45 pub id: String,
46 /// The repository this mirror belongs to.
47 pub repo_id: String,
48 /// Direction (push / pull).
49 pub direction: MirrorDirection,
50 /// The remote git URL.
51 pub remote_url: String,
52 /// Optional credential username.
53 pub username: Option<String>,
54 /// Optional credential password/token.
55 pub secret: Option<String>,
56 /// Optional branch filter (push only); blank means all branches.
57 pub branch_filter: Option<String>,
58 /// Sync interval in seconds; `0` means manual only.
59 pub interval_secs: i64,
60 /// Push mirrors: also sync right after a received push.
61 pub sync_on_push: bool,
62 /// When the last successful sync completed.
63 pub last_sync_at: Option<i64>,
64 /// The last sync error, if the most recent attempt failed.
65 pub last_error: Option<String>,
66 /// Creation time.
67 pub created_at: i64,
68}
69
70/// Fields to create a mirror.
71#[derive(Debug, Clone)]
72pub struct NewMirror {
73 /// The repository this mirror belongs to.
74 pub repo_id: String,
75 /// Direction (push / pull).
76 pub direction: MirrorDirection,
77 /// The remote git URL.
78 pub remote_url: String,
79 /// Optional credential username.
80 pub username: Option<String>,
81 /// Optional credential password/token.
82 pub secret: Option<String>,
83 /// Optional branch filter (push only).
84 pub branch_filter: Option<String>,
85 /// Sync interval in seconds (`0` = manual).
86 pub interval_secs: i64,
87 /// Push mirrors: sync after a received push.
88 pub sync_on_push: bool,
89}
90
91/// Columns of `mirrors`, in a fixed order shared by every `SELECT`.
92const MIRROR_COLUMNS: &str = "id, repo_id, direction, remote_url, username, secret, \
93 branch_filter, interval_secs, sync_on_push, last_sync_at, last_error, created_at";
94
95impl Store {
96 /// Map a `mirrors` row.
97 fn map_mirror(&self, row: &AnyRow) -> Result<Mirror, sqlx::Error> {
98 Ok(Mirror {
99 id: row.try_get("id")?,
100 repo_id: row.try_get("repo_id")?,
101 direction: MirrorDirection::parse(&row.try_get::<String, _>("direction")?),
102 remote_url: row.try_get("remote_url")?,
103 username: row.try_get("username")?,
104 secret: row.try_get("secret")?,
105 branch_filter: row.try_get("branch_filter")?,
106 interval_secs: row.try_get("interval_secs")?,
107 sync_on_push: self.get_bool(row, "sync_on_push")?,
108 last_sync_at: row.try_get("last_sync_at")?,
109 last_error: row.try_get("last_error")?,
110 created_at: row.try_get("created_at")?,
111 })
112 }
113
114 /// Create a mirror.
115 ///
116 /// # Errors
117 ///
118 /// Returns [`StoreError::Query`] on a database failure.
119 pub async fn create_mirror(&self, new: NewMirror) -> Result<Mirror, StoreError> {
120 let mirror = Mirror {
121 id: new_id(),
122 repo_id: new.repo_id,
123 direction: new.direction,
124 remote_url: new.remote_url,
125 username: new.username,
126 secret: new.secret,
127 branch_filter: new.branch_filter,
128 interval_secs: new.interval_secs,
129 sync_on_push: new.sync_on_push,
130 last_sync_at: None,
131 last_error: None,
132 created_at: now_ms(),
133 };
134 sqlx::query(
135 "INSERT INTO mirrors \
136 (id, repo_id, direction, remote_url, username, secret, branch_filter, \
137 interval_secs, sync_on_push, last_sync_at, last_error, created_at) \
138 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)",
139 )
140 .bind(&mirror.id)
141 .bind(&mirror.repo_id)
142 .bind(mirror.direction.as_str())
143 .bind(&mirror.remote_url)
144 .bind(&mirror.username)
145 .bind(&mirror.secret)
146 .bind(&mirror.branch_filter)
147 .bind(mirror.interval_secs)
148 .bind(mirror.sync_on_push)
149 .bind(mirror.last_sync_at)
150 .bind(&mirror.last_error)
151 .bind(mirror.created_at)
152 .execute(&self.pool)
153 .await?;
154 Ok(mirror)
155 }
156
157 /// List a repository's mirrors, newest first.
158 ///
159 /// # Errors
160 ///
161 /// Returns [`StoreError::Query`] on a database failure.
162 pub async fn mirrors_for_repo(&self, repo_id: &str) -> Result<Vec<Mirror>, StoreError> {
163 let sql = format!(
164 "SELECT {MIRROR_COLUMNS} FROM mirrors WHERE repo_id = $1 ORDER BY created_at DESC"
165 );
166 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
167 .bind(repo_id)
168 .fetch_all(&self.pool)
169 .await?;
170 rows.iter()
171 .map(|r| self.map_mirror(r))
172 .collect::<Result<_, _>>()
173 .map_err(Into::into)
174 }
175
176 /// Fetch a mirror by id.
177 ///
178 /// # Errors
179 ///
180 /// Returns [`StoreError::Query`] on a database failure.
181 pub async fn mirror_by_id(&self, id: &str) -> Result<Option<Mirror>, StoreError> {
182 let sql = format!("SELECT {MIRROR_COLUMNS} FROM mirrors WHERE id = $1");
183 let row = sqlx::query(sqlx::AssertSqlSafe(sql))
184 .bind(id)
185 .fetch_optional(&self.pool)
186 .await?;
187 Ok(row.as_ref().map(|r| self.map_mirror(r)).transpose()?)
188 }
189
190 /// Delete a mirror by id. Returns whether a row was removed.
191 ///
192 /// # Errors
193 ///
194 /// Returns [`StoreError::Query`] on a database failure.
195 pub async fn delete_mirror(&self, id: &str) -> Result<bool, StoreError> {
196 let n = sqlx::query("DELETE FROM mirrors WHERE id = $1")
197 .bind(id)
198 .execute(&self.pool)
199 .await?
200 .rows_affected();
201 Ok(n > 0)
202 }
203
204 /// Mirrors whose periodic sync is due at `now` (interval elapsed, or never
205 /// synced), across every repo. `interval_secs = 0` mirrors are excluded.
206 ///
207 /// # Errors
208 ///
209 /// Returns [`StoreError::Query`] on a database failure.
210 pub async fn due_mirrors(&self, now_ms: i64) -> Result<Vec<Mirror>, StoreError> {
211 let sql = format!(
212 "SELECT {MIRROR_COLUMNS} FROM mirrors \
213 WHERE interval_secs > 0 \
214 AND (last_sync_at IS NULL OR last_sync_at + interval_secs * 1000 <= $1) \
215 ORDER BY created_at ASC"
216 );
217 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
218 .bind(now_ms)
219 .fetch_all(&self.pool)
220 .await?;
221 rows.iter()
222 .map(|r| self.map_mirror(r))
223 .collect::<Result<_, _>>()
224 .map_err(Into::into)
225 }
226
227 /// Push mirrors for `repo_id` that should sync after a received push.
228 ///
229 /// # Errors
230 ///
231 /// Returns [`StoreError::Query`] on a database failure.
232 pub async fn push_mirrors_on_push(&self, repo_id: &str) -> Result<Vec<Mirror>, StoreError> {
233 let sql = format!(
234 "SELECT {MIRROR_COLUMNS} FROM mirrors \
235 WHERE repo_id = $1 AND direction = 'push' AND sync_on_push = $2"
236 );
237 let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
238 .bind(repo_id)
239 .bind(true)
240 .fetch_all(&self.pool)
241 .await?;
242 rows.iter()
243 .map(|r| self.map_mirror(r))
244 .collect::<Result<_, _>>()
245 .map_err(Into::into)
246 }
247
248 /// Record the outcome of a sync attempt: on success clear the error and stamp
249 /// `last_sync_at`; on failure store the error message.
250 ///
251 /// # Errors
252 ///
253 /// Returns [`StoreError::Query`] on a database failure.
254 pub async fn mirror_synced(&self, id: &str, error: Option<&str>) -> Result<(), StoreError> {
255 sqlx::query("UPDATE mirrors SET last_sync_at = $1, last_error = $2 WHERE id = $3")
256 .bind(now_ms())
257 .bind(error)
258 .bind(id)
259 .execute(&self.pool)
260 .await?;
261 Ok(())
262 }
263}