fabrica

hanna/fabrica

main
Switch branch
mainView all branches
Clone
HTTPS
SSH

feat(store): mirror configuration schema and queries

8978582 · hanna committed on 2026-07-26

Add the mirrors table (migration 0014) and store layer for repository
mirrors — push (out) and pull (in): create/list/delete, due_mirrors for
the periodic scheduler, push_mirrors_on_push for post-receive triggers,
and mirror_synced to record each attempt's outcome. Credentials are
stored reversibly since a mirror must authenticate to its remote.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed by hannaSSH key fingerprint: SHA256:4g9cWhkAAw8gwqhJUcVbSnGEvdGwklW+1Aa/fMUu59k
4 files changed · +301 −0UnifiedSplit
crates/store/src/lib.rs +2 −0
@@ -32,6 +32,7 @@ mod issues;
3232 mod keys;
3333 mod labels;
3434 mod lfs;
35+mod mirrors;
3536 mod repos;
3637 mod sessions;
3738 mod tokens;
@@ -49,6 +50,7 @@ pub use crate::admin::{AdminCounts, NewSignupInvite, SignupInvite};
4950 pub use crate::invites::NewInvite;
5051 pub use crate::issues::StateCounts;
5152 pub use crate::keys::NewKey;
53+pub use crate::mirrors::{Mirror, MirrorDirection, NewMirror};
5254 pub use crate::repos::{Collaborator, NewRepo};
5355 pub use crate::sessions::NewSession;
5456 pub use crate::tokens::NewToken;
crates/store/src/mirrors.rs +263 −0
@@ -0,0 +1,263 @@
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+
10+use sqlx::Row;
11+use sqlx::any::AnyRow;
12+
13+use 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)]
17+pub 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+
24+impl 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)]
43+pub 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)]
72+pub 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`.
92+const 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+
95+impl 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+}
migrations/postgres/0014_mirrors.sql +18 −0
@@ -0,0 +1,18 @@
1+-- Repository mirrors: outbound (push to a remote) and inbound (pull from a
2+-- remote). Credentials are stored so the periodic sync can authenticate; prefer
3+-- a scoped access token over a password.
4+CREATE TABLE mirrors (
5+ id TEXT PRIMARY KEY,
6+ repo_id TEXT NOT NULL REFERENCES repos(id) ON DELETE CASCADE,
7+ direction TEXT NOT NULL, -- 'push' | 'pull'
8+ remote_url TEXT NOT NULL,
9+ username TEXT,
10+ secret TEXT, -- password / token (reversible by design)
11+ branch_filter TEXT, -- push only; blank = all branches
12+ interval_secs BIGINT NOT NULL, -- 0 = manual sync only
13+ sync_on_push INTEGER NOT NULL, -- push mirrors: sync after a received push
14+ last_sync_at BIGINT,
15+ last_error TEXT,
16+ created_at BIGINT NOT NULL
17+);
18+CREATE INDEX idx_mirrors_repo ON mirrors (repo_id);
migrations/sqlite/0014_mirrors.sql +18 −0
@@ -0,0 +1,18 @@
1+-- Repository mirrors: outbound (push to a remote) and inbound (pull from a
2+-- remote). Credentials are stored so the periodic sync can authenticate; prefer
3+-- a scoped access token over a password.
4+CREATE TABLE mirrors (
5+ id TEXT PRIMARY KEY,
6+ repo_id TEXT NOT NULL REFERENCES repos(id) ON DELETE CASCADE,
7+ direction TEXT NOT NULL, -- 'push' | 'pull'
8+ remote_url TEXT NOT NULL,
9+ username TEXT,
10+ secret TEXT, -- password / token (reversible by design)
11+ branch_filter TEXT, -- push only; blank = all branches
12+ interval_secs BIGINT NOT NULL, -- 0 = manual sync only
13+ sync_on_push INTEGER NOT NULL, -- push mirrors: sync after a received push
14+ last_sync_at BIGINT,
15+ last_error TEXT,
16+ created_at BIGINT NOT NULL
17+);
18+CREATE INDEX idx_mirrors_repo ON mirrors (repo_id);