crates/hub/src/db/repos.rs at api-unknown-fields · andymccall/oxygenhub

◆ api-unknown-fields oxygenhub / crates / hub / src / db / repos.rs

20.3 KiB · Rust · last changed Comments on drops (refs #280) 14 days ago source blame history raw

//! The `repositories` and `pushes` tables.

use super::{now_ms, Pool};
use hub_api::Visibility;
use sqlx::FromRow;

#[derive(Clone, Debug, Default, FromRow)]
pub struct RepoRow {
    pub id: i64,
    pub owner: String,
    pub name: String,
    pub description: String,
    pub default_bookmark: String,
    pub visibility: String,
    pub created_ms: i64,
    pub updated_ms: i64,
    #[sqlx(default)]
    pub git_mirror: bool,
    #[sqlx(default)]
    pub git_mirror_state: String,
    /// The repository this one was forked from (migration 0009).
    #[sqlx(default)]
    pub forked_from: Option<i64>,
    /// Space-separated topics (migration 0010).
    #[sqlx(default)]
    pub topics: String,
    /// OxygenFlow: `on`, `off`, or empty to follow hub.toml (migration 0018).
    #[sqlx(default)]
    pub flow: String,
    /// Its jobs' priority: `low`, `normal`, `high`, or empty to follow its
    /// organisation's and the site's (oxygenhub#244, migration 0073).
    #[sqlx(default)]
    pub flow_priority: String,
    /// Diffusion: `off`, `green`, `any`, or empty to follow hub.toml
    /// (migration 0058, #148).
    #[sqlx(default)]
    pub diffusion: String,
    /// Diffusion auto-land: `auto`, `propose`, or empty for propose
    /// (migration 0061).
    #[sqlx(default)]
    pub diffusion_land: String,
    /// Backups (migration 0062): empty with its owner's, `off`, or its own
    /// `hourly` or `daily`.
    #[sqlx(default)]
    pub backup: String,
    /// Read-only, with why (migration 0040, #108).
    #[sqlx(default)]
    pub archived: bool,
    #[sqlx(default)]
    pub archived_reason: String,
    /// Agents' tokens may approve or reject its approval gates (#258).
    #[sqlx(default)]
    pub agents_decide_gates: bool,
    /// Newest run wins (#267): a superseded run already running finishes;
    /// only queued ones are cancelled. Off by default.
    #[sqlx(default)]
    pub flow_keep_running: bool,
    /// Where a push may skip its run with `[skip flow]` (#284, migration
    /// 0088): `any` bookmark, `others` than the default one, or `never`.
    #[sqlx(default)]
    pub flow_skip: String,
    /// A push run reuses a job's pass on the same state (#265, migration
    /// 0090). On by default.
    #[sqlx(default)]
    pub flow_reuse: bool,
    /// Pushes that add possible secrets are refused (#268, migration
    /// 0085). On by default.
    #[sqlx(default)]
    pub push_protection: bool,
    /// `repo`, or `drop` for a drop (#279, migration 0087): a small
    /// repository named `drops/ID` that no repository list shows.
    #[sqlx(default)]
    pub kind: String,
    /// A drop's title.
    #[sqlx(default)]
    pub title: String,
    /// Counts for the tabs; filled by `get` only.
    #[sqlx(default)]
    pub open_issues: i64,
    #[sqlx(default)]
    pub open_proposals: i64,
}

impl RepoRow {
    pub fn visibility(&self) -> Visibility {
        Visibility::parse(&self.visibility).unwrap_or(Visibility::Private)
    }

    pub fn topic_list(&self) -> Vec<&str> {
        self.topics.split_whitespace().collect()
    }

    pub fn slug(&self) -> String {
        format!("{}/{}", self.owner, self.name)
    }

    /// A drop (#279), not a repository.
    pub fn is_drop(&self) -> bool {
        self.kind == DROP
    }
}

/// `repositories.kind` of a drop (#279).
pub const DROP: &str = "drop";

#[derive(Clone, Debug, FromRow)]
pub struct PushRow {
    pub id: i64,
    pub at_ms: i64,
    pub reference: String,
    pub ok: bool,
    pub reason: String,
    pub old_state: String,
    pub new_state: String,
    pub heads: i64,
    pub changes: i64,
    pub pusher: String,
}

/// One reference's outcome in a push, as the protocol backend reports it.
#[derive(Clone, Debug)]
pub struct PushOutcome {
    pub reference: String,
    pub ok: bool,
    pub reason: String,
    pub old_state: String,
    pub new_state: String,
    pub heads: i64,
    pub changes: i64,
}

/// Every repository, drops left out: what every list, search and count
/// of repositories starts from (#279).
pub async fn list(pool: &Pool) -> sqlx::Result<Vec<RepoRow>> {
    sqlx::query_as("SELECT * FROM repositories WHERE kind = 'repo' ORDER BY updated_ms DESC, owner, name").fetch_all(pool).await
}

/// Every repository and every drop: for upkeep that covers all stored
/// repositories (gc, the history scan, `oxygenhub check`), never a list
/// people see unless it labels the drops.
pub async fn list_with_drops(pool: &Pool) -> sqlx::Result<Vec<RepoRow>> {
    sqlx::query_as("SELECT * FROM repositories ORDER BY updated_ms DESC, owner, name").fetch_all(pool).await
}

pub async fn get_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<RepoRow>> {
    sqlx::query_as("SELECT * FROM repositories WHERE id = $1").bind(id).fetch_optional(pool).await
}

pub async fn list_by_owner(pool: &Pool, owner: &str) -> sqlx::Result<Vec<RepoRow>> {
    sqlx::query_as("SELECT * FROM repositories WHERE owner = $1 AND kind = 'repo' ORDER BY updated_ms DESC, name")
        .bind(owner)
        .fetch_all(pool)
        .await
}

// ------------------------------------------------------------------ drops

/// A drop's row (#279): a repository named `drops/ID`, its kind `drop`,
/// push protection on and Flow off. `visibility` is `public`, or
/// `private` for a secret drop.
pub async fn create_drop(
    pool: &Pool,
    owner: &str,
    id: &str,
    title: &str,
    description: &str,
    visibility: Visibility,
) -> sqlx::Result<RepoRow> {
    let now = now_ms();
    sqlx::query_as(
        "INSERT INTO repositories (owner, name, description, visibility, created_ms, updated_ms, kind, title, flow) \
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'off') RETURNING *",
    )
    .bind(owner)
    .bind(format!("{}/{id}", crate::repos::names::DROPS))
    .bind(description)
    .bind(visibility.as_str())
    .bind(now)
    .bind(now)
    .bind(DROP)
    .bind(title)
    .fetch_one(pool)
    .await
}

/// A comment on a drop (#280).
#[derive(Clone, Debug, FromRow)]
pub struct DropComment {
    pub id: i64,
    pub repo_id: i64,
    pub author_id: Option<i64>,
    pub author: String,
    pub body: String,
    pub created_ms: i64,
}

pub async fn add_drop_comment(pool: &Pool, repo_id: i64, author_id: i64, author: &str, body: &str) -> sqlx::Result<i64> {
    sqlx::query_scalar("INSERT INTO drop_comments (repo_id, author_id, author, body, created_ms) VALUES ($1, $2, $3, $4, $5) RETURNING id")
        .bind(repo_id)
        .bind(author_id)
        .bind(author)
        .bind(body)
        .bind(now_ms())
        .fetch_one(pool)
        .await
}

/// A drop's comments, oldest first.
pub async fn drop_comments(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<DropComment>> {
    sqlx::query_as("SELECT * FROM drop_comments WHERE repo_id = $1 ORDER BY created_ms, id").bind(repo_id).fetch_all(pool).await
}

pub async fn drop_comment(pool: &Pool, repo_id: i64, id: i64) -> sqlx::Result<Option<DropComment>> {
    sqlx::query_as("SELECT * FROM drop_comments WHERE repo_id = $1 AND id = $2").bind(repo_id).bind(id).fetch_optional(pool).await
}

pub async fn delete_drop_comment(pool: &Pool, repo_id: i64, id: i64) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM drop_comments WHERE repo_id = $1 AND id = $2").bind(repo_id).bind(id).execute(pool).await?;
    Ok(())
}

/// An owner's drops, newest activity first.
pub async fn drops_of(pool: &Pool, owner: &str) -> sqlx::Result<Vec<RepoRow>> {
    sqlx::query_as("SELECT * FROM repositories WHERE kind = 'drop' AND owner = $1 ORDER BY updated_ms DESC, id DESC")
        .bind(owner)
        .fetch_all(pool)
        .await
}

/// The drop with this id, whoever owns it: ids are random and made
/// unique across the hub.
pub async fn drop_by_id(pool: &Pool, id: &str) -> sqlx::Result<Option<RepoRow>> {
    sqlx::query_as("SELECT * FROM repositories WHERE kind = 'drop' AND name = $1")
        .bind(format!("{}/{id}", crate::repos::names::DROPS))
        .fetch_optional(pool)
        .await
}

/// A drop's title, description and visibility, and that it changed.
pub async fn set_drop(pool: &Pool, row_id: i64, title: &str, description: &str, visibility: Visibility) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET title = $1, description = $2, visibility = $3, updated_ms = $4 WHERE id = $5 AND kind = 'drop'")
        .bind(title)
        .bind(description)
        .bind(visibility.as_str())
        .bind(now_ms())
        .bind(row_id)
        .execute(pool)
        .await?;
    Ok(())
}

/// One repository, with its open issue and proposal counts for the tabs.
pub async fn get(pool: &Pool, owner: &str, name: &str) -> sqlx::Result<Option<RepoRow>> {
    sqlx::query_as(
        "SELECT r.*, \
           (SELECT COUNT(*) FROM issues i WHERE i.repo_id = r.id AND i.state = 'open') AS open_issues, \
           (SELECT COUNT(*) FROM proposals p WHERE p.repo_id = r.id AND p.state IN ('open', 'draft')) AS open_proposals \
         FROM repositories r WHERE r.owner = $1 AND r.name = $2",
    )
    .bind(owner)
    .bind(name)
    .fetch_optional(pool)
    .await
}

/// Create a repository row. One under an organisation of two or more
/// members requires one sealed approval to land a proposal (oxygenhub#64),
/// since someone other than the author can then give it; others require none.
pub async fn create(pool: &Pool, owner: &str, name: &str, description: &str, visibility: Visibility) -> sqlx::Result<RepoRow> {
    let now = now_ms();
    sqlx::query_as(
        "INSERT INTO repositories (owner, name, description, visibility, created_ms, updated_ms, required_approvals) \
         VALUES ($1, $2, $3, $4, $5, $6, (SELECT CASE WHEN COUNT(*) >= 2 THEN 1 ELSE 0 END FROM org_members m JOIN organisations o ON o.id = m.org_id WHERE o.name = $7)) RETURNING *",
    )
    .bind(owner)
    .bind(name)
    .bind(description)
    .bind(visibility.as_str())
    .bind(now)
    .bind(now)
    .bind(owner)
    .fetch_one(pool)
    .await
}

/// The repository an old name now points at, after a transfer.
pub async fn moved_to(pool: &Pool, owner: &str, name: &str) -> sqlx::Result<Option<RepoRow>> {
    let id: Option<i64> = sqlx::query_scalar("SELECT repo_id FROM repo_redirects WHERE old_owner = $1 AND old_name = $2")
        .bind(owner)
        .bind(name)
        .fetch_optional(pool)
        .await?;
    match id {
        Some(id) => get_by_id(pool, id).await,
        None => Ok(None),
    }
}

/// Move a repository to another owner, keeping its name: the row, and a
/// redirect from the old name. A redirect that pointed at the new name
/// goes, so names never loop.
pub async fn transfer(pool: &Pool, id: i64, old_owner: &str, name: &str, new_owner: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET owner = $1, updated_ms = $2 WHERE id = $3")
        .bind(new_owner)
        .bind(now_ms())
        .bind(id)
        .execute(pool)
        .await?;
    sqlx::query("DELETE FROM repo_redirects WHERE old_owner = $1 AND old_name = $2").bind(new_owner).bind(name).execute(pool).await?;
    sqlx::query("INSERT INTO repo_redirects (old_owner, old_name, repo_id, moved_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (old_owner, old_name) DO UPDATE SET repo_id = excluded.repo_id, moved_ms = excluded.moved_ms")
        .bind(old_owner)
        .bind(name)
        .bind(id)
        .bind(now_ms())
        .execute(pool)
        .await?;
    Ok(())
}

/// Archive (read-only) or bring back a repository.
pub async fn set_archived(pool: &Pool, id: i64, archived: bool, reason: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET archived = $1, archived_reason = $2, updated_ms = $3 WHERE id = $4")
        .bind(archived)
        .bind(if archived { reason } else { "" })
        .bind(now_ms())
        .bind(id)
        .execute(pool)
        .await?;
    Ok(())
}

pub async fn delete(pool: &Pool, id: i64) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM repositories WHERE id = $1").bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_description(pool: &Pool, id: i64, description: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET description = $1 WHERE id = $2").bind(description).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_default_bookmark(pool: &Pool, id: i64, bookmark: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET default_bookmark = $1 WHERE id = $2").bind(bookmark).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_visibility(pool: &Pool, id: i64, visibility: Visibility) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET visibility = $1 WHERE id = $2").bind(visibility.as_str()).bind(id).execute(pool).await?;
    Ok(())
}

/// Record a push's per-reference results and bump the repository's activity.
pub async fn record_push(pool: &Pool, id: i64, pusher: Option<(i64, &str)>, results: &[PushOutcome]) -> sqlx::Result<()> {
    record_push_via(pool, id, pusher, "", results).await
}

/// [`record_push`], naming the agent token the push came through (#217).
pub async fn record_push_via(pool: &Pool, id: i64, pusher: Option<(i64, &str)>, via: &str, results: &[PushOutcome]) -> sqlx::Result<()> {
    let now = now_ms();
    let mut tx = super::begin_write(pool).await?;
    for r in results {
        sqlx::query("INSERT INTO pushes (repo_id, at_ms, reference, ok, reason, old_state, new_state, heads, changes, account_id, pusher, via) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)")
            .bind(id)
            .bind(now)
            .bind(&r.reference)
            .bind(r.ok)
            .bind(&r.reason)
            .bind(&r.old_state)
            .bind(&r.new_state)
            .bind(r.heads)
            .bind(r.changes)
            .bind(pusher.map(|p| p.0).filter(|id| *id > 0))
            .bind(pusher.map(|p| p.1).unwrap_or(""))
            .bind(via)
            .execute(&mut *tx)
            .await?;
    }
    if results.iter().any(|r| r.ok) {
        sqlx::query("UPDATE repositories SET updated_ms = $1 WHERE id = $2").bind(now).bind(id).execute(&mut *tx).await?;
    }
    tx.commit().await
}

pub async fn pushes(pool: &Pool, id: i64, limit: i64) -> sqlx::Result<Vec<PushRow>> {
    sqlx::query_as("SELECT id, at_ms, reference, ok, reason, old_state, new_state, heads, changes, pusher FROM pushes WHERE repo_id = $1 ORDER BY at_ms DESC, id DESC LIMIT $2")
        .bind(id)
        .bind(limit)
        .fetch_all(pool)
        .await
}

/// Topics from user input: comma or space separated, lower-cased,
/// letters, digits and dashes, at most 35 characters each and 20 in all.
/// Returns the stored form or what is wrong.
pub fn parse_topics(input: &str) -> Result<String, String> {
    let mut out: Vec<String> = Vec::new();
    for t in input.split([',', ' ', '\n', '\t']).map(|t| t.trim().to_ascii_lowercase()).filter(|t| !t.is_empty()) {
        if t.len() > 35 || !t.chars().all(|c| c.is_ascii_alphanumeric() || c == '-') || t.starts_with('-') {
            return Err(format!("\"{t}\" is not a topic: use letters, digits and dashes, up to 35 characters"));
        }
        if !out.contains(&t) {
            out.push(t);
        }
    }
    if out.len() > 20 {
        return Err("At most 20 topics.".into());
    }
    Ok(out.join(" "))
}

pub async fn set_topics(pool: &Pool, id: i64, topics: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET topics = $1 WHERE id = $2").bind(topics).bind(id).execute(pool).await?;
    Ok(())
}

/// Star or unstar; returns whether it is now starred.
pub async fn toggle_star(pool: &Pool, account_id: i64, repo_id: i64) -> sqlx::Result<bool> {
    let gone = sqlx::query("DELETE FROM stars WHERE account_id = $1 AND repo_id = $2").bind(account_id).bind(repo_id).execute(pool).await?;
    if gone.rows_affected() > 0 {
        return Ok(false);
    }
    sqlx::query("INSERT INTO stars (account_id, repo_id, at_ms) VALUES ($1, $2, $3)")
        .bind(account_id)
        .bind(repo_id)
        .bind(now_ms())
        .execute(pool)
        .await?;
    Ok(true)
}

pub async fn starred(pool: &Pool, account_id: i64, repo_id: i64) -> sqlx::Result<bool> {
    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stars WHERE account_id = $1 AND repo_id = $2")
        .bind(account_id)
        .bind(repo_id)
        .fetch_one(pool)
        .await?;
    Ok(n > 0)
}

/// Stars and watchers of one repository.
pub async fn counts(pool: &Pool, repo_id: i64) -> sqlx::Result<(i64, i64)> {
    let stars: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stars WHERE repo_id = $1").bind(repo_id).fetch_one(pool).await?;
    let watchers = super::notify::watchers(pool, repo_id).await?.len() as i64;
    Ok((stars, watchers))
}

/// Star counts of every starred repository.
pub async fn star_counts(pool: &Pool) -> sqlx::Result<std::collections::HashMap<i64, i64>> {
    let rows: Vec<(i64, i64)> = sqlx::query_as("SELECT repo_id, COUNT(*) FROM stars GROUP BY repo_id").fetch_all(pool).await?;
    Ok(rows.into_iter().collect())
}

/// Record that `id` is a fork of `from`.
pub async fn set_forked_from(pool: &Pool, id: i64, from: i64) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET forked_from = $1 WHERE id = $2").bind(from).bind(id).execute(pool).await?;
    Ok(())
}

/// The forks of `id`, newest activity first.
pub async fn forks_of(pool: &Pool, id: i64) -> sqlx::Result<Vec<RepoRow>> {
    sqlx::query_as("SELECT * FROM repositories WHERE forked_from = $1 AND kind = 'repo' ORDER BY updated_ms DESC")
        .bind(id)
        .fetch_all(pool)
        .await
}

/// `on`, `off`, or `` (follow hub.toml's `[flow] repos`).
pub async fn set_flow(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET flow = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_flow_priority(pool: &Pool, id: i64, priority: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET flow_priority = $1 WHERE id = $2").bind(priority).bind(id).execute(pool).await?;
    Ok(())
}

/// Whether agents' tokens may decide the repository's approval gates (#258).
pub async fn set_agents_decide_gates(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET agents_decide_gates = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
    Ok(())
}

/// Whether a new run leaves the older runs it supersedes that are already
/// running to finish (#267), cancelling only the queued ones.
pub async fn set_flow_reuse(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET flow_reuse = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_flow_skip(pool: &Pool, id: i64, skip: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET flow_skip = $1 WHERE id = $2").bind(skip).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_flow_keep_running(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET flow_keep_running = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_backup(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET backup = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_diffusion_land(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET diffusion_land = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_diffusion(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET diffusion = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_git_mirror(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET git_mirror = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
    Ok(())
}

pub async fn set_git_mirror_state(pool: &Pool, id: i64, state: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE repositories SET git_mirror_state = $1 WHERE id = $2").bind(state).bind(id).execute(pool).await?;
    Ok(())
}

#[cfg(test)]
mod topic_tests {
    use super::parse_topics;

    #[test]
    fn topics_are_normalised_and_checked() {
        assert_eq!(parse_topics("Rust, vcs  retro-computing,rust").unwrap(), "rust vcs retro-computing");
        assert_eq!(parse_topics("").unwrap(), "");
        assert!(parse_topics("c++").is_err());
        assert!(parse_topics("-lead").is_err());
        assert!(parse_topics(&"x".repeat(36)).is_err());
        assert!(parse_topics(&(0..21).map(|i| format!("t{i}")).collect::<Vec<_>>().join(" ")).is_err());
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1	//! The `repositories` and `pushes` tables.
2
3	use super::{now_ms, Pool};
4	use hub_api::Visibility;
5	use sqlx::FromRow;
6
7	#[derive(Clone, Debug, Default, FromRow)]
8	pub struct RepoRow {
9	pub id: i64,
10	pub owner: String,
11	pub name: String,
12	pub description: String,
13	pub default_bookmark: String,
14	pub visibility: String,
15	pub created_ms: i64,
16	pub updated_ms: i64,
17	#[sqlx(default)]
18	pub git_mirror: bool,
19	#[sqlx(default)]
20	pub git_mirror_state: String,
21	/// The repository this one was forked from (migration 0009).
22	#[sqlx(default)]
23	pub forked_from: Option<i64>,
24	/// Space-separated topics (migration 0010).
25	#[sqlx(default)]
26	pub topics: String,
27	/// OxygenFlow: `on`, `off`, or empty to follow hub.toml (migration 0018).
28	#[sqlx(default)]
29	pub flow: String,
30	/// Its jobs' priority: `low`, `normal`, `high`, or empty to follow its
31	/// organisation's and the site's (oxygenhub#244, migration 0073).
32	#[sqlx(default)]
33	pub flow_priority: String,
34	/// Diffusion: `off`, `green`, `any`, or empty to follow hub.toml
35	/// (migration 0058, #148).
36	#[sqlx(default)]
37	pub diffusion: String,
38	/// Diffusion auto-land: `auto`, `propose`, or empty for propose
39	/// (migration 0061).
40	#[sqlx(default)]
41	pub diffusion_land: String,
42	/// Backups (migration 0062): empty with its owner's, `off`, or its own
43	/// `hourly` or `daily`.
44	#[sqlx(default)]
45	pub backup: String,
46	/// Read-only, with why (migration 0040, #108).
47	#[sqlx(default)]
48	pub archived: bool,
49	#[sqlx(default)]
50	pub archived_reason: String,
51	/// Agents' tokens may approve or reject its approval gates (#258).
52	#[sqlx(default)]
53	pub agents_decide_gates: bool,
54	/// Newest run wins (#267): a superseded run already running finishes;
55	/// only queued ones are cancelled. Off by default.
56	#[sqlx(default)]
57	pub flow_keep_running: bool,
58	/// Where a push may skip its run with `[skip flow]` (#284, migration
59	/// 0088): `any` bookmark, `others` than the default one, or `never`.
60	#[sqlx(default)]
61	pub flow_skip: String,
62	/// A push run reuses a job's pass on the same state (#265, migration
63	/// 0090). On by default.
64	#[sqlx(default)]
65	pub flow_reuse: bool,
66	/// Pushes that add possible secrets are refused (#268, migration
67	/// 0085). On by default.
68	#[sqlx(default)]
69	pub push_protection: bool,
70	/// `repo`, or `drop` for a drop (#279, migration 0087): a small
71	/// repository named `drops/ID` that no repository list shows.
72	#[sqlx(default)]
73	pub kind: String,
74	/// A drop's title.
75	#[sqlx(default)]
76	pub title: String,
77	/// Counts for the tabs; filled by `get` only.
78	#[sqlx(default)]
79	pub open_issues: i64,
80	#[sqlx(default)]
81	pub open_proposals: i64,
82	}
83
84	impl RepoRow {
85	pub fn visibility(&self) -> Visibility {
86	Visibility::parse(&self.visibility).unwrap_or(Visibility::Private)
87	}
88
89	pub fn topic_list(&self) -> Vec<&str> {
90	self.topics.split_whitespace().collect()
91	}
92
93	pub fn slug(&self) -> String {
94	format!("{}/{}", self.owner, self.name)
95	}
96
97	/// A drop (#279), not a repository.
98	pub fn is_drop(&self) -> bool {
99	self.kind == DROP
100	}
101	}
102
103	/// `repositories.kind` of a drop (#279).
104	pub const DROP: &str = "drop";
105
106	#[derive(Clone, Debug, FromRow)]
107	pub struct PushRow {
108	pub id: i64,
109	pub at_ms: i64,
110	pub reference: String,
111	pub ok: bool,
112	pub reason: String,
113	pub old_state: String,
114	pub new_state: String,
115	pub heads: i64,
116	pub changes: i64,
117	pub pusher: String,
118	}
119
120	/// One reference's outcome in a push, as the protocol backend reports it.
121	#[derive(Clone, Debug)]
122	pub struct PushOutcome {
123	pub reference: String,
124	pub ok: bool,
125	pub reason: String,
126	pub old_state: String,
127	pub new_state: String,
128	pub heads: i64,
129	pub changes: i64,
130	}
131
132	/// Every repository, drops left out: what every list, search and count
133	/// of repositories starts from (#279).
134	pub async fn list(pool: &Pool) -> sqlx::Result<Vec<RepoRow>> {
135	sqlx::query_as("SELECT * FROM repositories WHERE kind = 'repo' ORDER BY updated_ms DESC, owner, name").fetch_all(pool).await
136	}
137
138	/// Every repository and every drop: for upkeep that covers all stored
139	/// repositories (gc, the history scan, `oxygenhub check`), never a list
140	/// people see unless it labels the drops.
141	pub async fn list_with_drops(pool: &Pool) -> sqlx::Result<Vec<RepoRow>> {
142	sqlx::query_as("SELECT * FROM repositories ORDER BY updated_ms DESC, owner, name").fetch_all(pool).await
143	}
144
145	pub async fn get_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<RepoRow>> {
146	sqlx::query_as("SELECT * FROM repositories WHERE id = $1").bind(id).fetch_optional(pool).await
147	}
148
149	pub async fn list_by_owner(pool: &Pool, owner: &str) -> sqlx::Result<Vec<RepoRow>> {
150	sqlx::query_as("SELECT * FROM repositories WHERE owner = $1 AND kind = 'repo' ORDER BY updated_ms DESC, name")
151	.bind(owner)
152	.fetch_all(pool)
153	.await
154	}
155
156	// ------------------------------------------------------------------ drops
157
158	/// A drop's row (#279): a repository named `drops/ID`, its kind `drop`,
159	/// push protection on and Flow off. `visibility` is `public`, or
160	/// `private` for a secret drop.
161	pub async fn create_drop(
162	pool: &Pool,
163	owner: &str,
164	id: &str,
165	title: &str,
166	description: &str,
167	visibility: Visibility,
168	) -> sqlx::Result<RepoRow> {
169	let now = now_ms();
170	sqlx::query_as(
171	"INSERT INTO repositories (owner, name, description, visibility, created_ms, updated_ms, kind, title, flow) \
172	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'off') RETURNING *",
173	)
174	.bind(owner)
175	.bind(format!("{}/{id}", crate::repos::names::DROPS))
176	.bind(description)
177	.bind(visibility.as_str())
178	.bind(now)
179	.bind(now)
180	.bind(DROP)
181	.bind(title)
182	.fetch_one(pool)
183	.await
184	}
185
186	/// A comment on a drop (#280).
187	#[derive(Clone, Debug, FromRow)]
188	pub struct DropComment {
189	pub id: i64,
190	pub repo_id: i64,
191	pub author_id: Option<i64>,
192	pub author: String,
193	pub body: String,
194	pub created_ms: i64,
195	}
196
197	pub async fn add_drop_comment(pool: &Pool, repo_id: i64, author_id: i64, author: &str, body: &str) -> sqlx::Result<i64> {
198	sqlx::query_scalar("INSERT INTO drop_comments (repo_id, author_id, author, body, created_ms) VALUES ($1, $2, $3, $4, $5) RETURNING id")
199	.bind(repo_id)
200	.bind(author_id)
201	.bind(author)
202	.bind(body)
203	.bind(now_ms())
204	.fetch_one(pool)
205	.await
206	}
207
208	/// A drop's comments, oldest first.
209	pub async fn drop_comments(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<DropComment>> {
210	sqlx::query_as("SELECT * FROM drop_comments WHERE repo_id = $1 ORDER BY created_ms, id").bind(repo_id).fetch_all(pool).await
211	}
212
213	pub async fn drop_comment(pool: &Pool, repo_id: i64, id: i64) -> sqlx::Result<Option<DropComment>> {
214	sqlx::query_as("SELECT * FROM drop_comments WHERE repo_id = $1 AND id = $2").bind(repo_id).bind(id).fetch_optional(pool).await
215	}
216
217	pub async fn delete_drop_comment(pool: &Pool, repo_id: i64, id: i64) -> sqlx::Result<()> {
218	sqlx::query("DELETE FROM drop_comments WHERE repo_id = $1 AND id = $2").bind(repo_id).bind(id).execute(pool).await?;
219	Ok(())
220	}
221
222	/// An owner's drops, newest activity first.
223	pub async fn drops_of(pool: &Pool, owner: &str) -> sqlx::Result<Vec<RepoRow>> {
224	sqlx::query_as("SELECT * FROM repositories WHERE kind = 'drop' AND owner = $1 ORDER BY updated_ms DESC, id DESC")
225	.bind(owner)
226	.fetch_all(pool)
227	.await
228	}
229
230	/// The drop with this id, whoever owns it: ids are random and made
231	/// unique across the hub.
232	pub async fn drop_by_id(pool: &Pool, id: &str) -> sqlx::Result<Option<RepoRow>> {
233	sqlx::query_as("SELECT * FROM repositories WHERE kind = 'drop' AND name = $1")
234	.bind(format!("{}/{id}", crate::repos::names::DROPS))
235	.fetch_optional(pool)
236	.await
237	}
238
239	/// A drop's title, description and visibility, and that it changed.
240	pub async fn set_drop(pool: &Pool, row_id: i64, title: &str, description: &str, visibility: Visibility) -> sqlx::Result<()> {
241	sqlx::query("UPDATE repositories SET title = $1, description = $2, visibility = $3, updated_ms = $4 WHERE id = $5 AND kind = 'drop'")
242	.bind(title)
243	.bind(description)
244	.bind(visibility.as_str())
245	.bind(now_ms())
246	.bind(row_id)
247	.execute(pool)
248	.await?;
249	Ok(())
250	}
251
252	/// One repository, with its open issue and proposal counts for the tabs.
253	pub async fn get(pool: &Pool, owner: &str, name: &str) -> sqlx::Result<Option<RepoRow>> {
254	sqlx::query_as(
255	"SELECT r.*, \
256	(SELECT COUNT(*) FROM issues i WHERE i.repo_id = r.id AND i.state = 'open') AS open_issues, \
257	(SELECT COUNT(*) FROM proposals p WHERE p.repo_id = r.id AND p.state IN ('open', 'draft')) AS open_proposals \
258	FROM repositories r WHERE r.owner = $1 AND r.name = $2",
259	)
260	.bind(owner)
261	.bind(name)
262	.fetch_optional(pool)
263	.await
264	}
265
266	/// Create a repository row. One under an organisation of two or more
267	/// members requires one sealed approval to land a proposal (oxygenhub#64),
268	/// since someone other than the author can then give it; others require none.
269	pub async fn create(pool: &Pool, owner: &str, name: &str, description: &str, visibility: Visibility) -> sqlx::Result<RepoRow> {
270	let now = now_ms();
271	sqlx::query_as(
272	"INSERT INTO repositories (owner, name, description, visibility, created_ms, updated_ms, required_approvals) \
273	VALUES ($1, $2, $3, $4, $5, $6, (SELECT CASE WHEN COUNT() >= 2 THEN 1 ELSE 0 END FROM org_members m JOIN organisations o ON o.id = m.org_id WHERE o.name = $7)) RETURNING ",
274	)
275	.bind(owner)
276	.bind(name)
277	.bind(description)
278	.bind(visibility.as_str())
279	.bind(now)
280	.bind(now)
281	.bind(owner)
282	.fetch_one(pool)
283	.await
284	}
285
286	/// The repository an old name now points at, after a transfer.
287	pub async fn moved_to(pool: &Pool, owner: &str, name: &str) -> sqlx::Result<Option<RepoRow>> {
288	let id: Option<i64> = sqlx::query_scalar("SELECT repo_id FROM repo_redirects WHERE old_owner = $1 AND old_name = $2")
289	.bind(owner)
290	.bind(name)
291	.fetch_optional(pool)
292	.await?;
293	match id {
294	Some(id) => get_by_id(pool, id).await,
295	None => Ok(None),
296	}
297	}
298
299	/// Move a repository to another owner, keeping its name: the row, and a
300	/// redirect from the old name. A redirect that pointed at the new name
301	/// goes, so names never loop.
302	pub async fn transfer(pool: &Pool, id: i64, old_owner: &str, name: &str, new_owner: &str) -> sqlx::Result<()> {
303	sqlx::query("UPDATE repositories SET owner = $1, updated_ms = $2 WHERE id = $3")
304	.bind(new_owner)
305	.bind(now_ms())
306	.bind(id)
307	.execute(pool)
308	.await?;
309	sqlx::query("DELETE FROM repo_redirects WHERE old_owner = $1 AND old_name = $2").bind(new_owner).bind(name).execute(pool).await?;
310	sqlx::query("INSERT INTO repo_redirects (old_owner, old_name, repo_id, moved_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (old_owner, old_name) DO UPDATE SET repo_id = excluded.repo_id, moved_ms = excluded.moved_ms")
311	.bind(old_owner)
312	.bind(name)
313	.bind(id)
314	.bind(now_ms())
315	.execute(pool)
316	.await?;
317	Ok(())
318	}
319
320	/// Archive (read-only) or bring back a repository.
321	pub async fn set_archived(pool: &Pool, id: i64, archived: bool, reason: &str) -> sqlx::Result<()> {
322	sqlx::query("UPDATE repositories SET archived = $1, archived_reason = $2, updated_ms = $3 WHERE id = $4")
323	.bind(archived)
324	.bind(if archived { reason } else { "" })
325	.bind(now_ms())
326	.bind(id)
327	.execute(pool)
328	.await?;
329	Ok(())
330	}
331
332	pub async fn delete(pool: &Pool, id: i64) -> sqlx::Result<()> {
333	sqlx::query("DELETE FROM repositories WHERE id = $1").bind(id).execute(pool).await?;
334	Ok(())
335	}
336
337	pub async fn set_description(pool: &Pool, id: i64, description: &str) -> sqlx::Result<()> {
338	sqlx::query("UPDATE repositories SET description = $1 WHERE id = $2").bind(description).bind(id).execute(pool).await?;
339	Ok(())
340	}
341
342	pub async fn set_default_bookmark(pool: &Pool, id: i64, bookmark: &str) -> sqlx::Result<()> {
343	sqlx::query("UPDATE repositories SET default_bookmark = $1 WHERE id = $2").bind(bookmark).bind(id).execute(pool).await?;
344	Ok(())
345	}
346
347	pub async fn set_visibility(pool: &Pool, id: i64, visibility: Visibility) -> sqlx::Result<()> {
348	sqlx::query("UPDATE repositories SET visibility = $1 WHERE id = $2").bind(visibility.as_str()).bind(id).execute(pool).await?;
349	Ok(())
350	}
351
352	/// Record a push's per-reference results and bump the repository's activity.
353	pub async fn record_push(pool: &Pool, id: i64, pusher: Option<(i64, &str)>, results: &[PushOutcome]) -> sqlx::Result<()> {
354	record_push_via(pool, id, pusher, "", results).await
355	}
356
357	/// [`record_push`], naming the agent token the push came through (#217).
358	pub async fn record_push_via(pool: &Pool, id: i64, pusher: Option<(i64, &str)>, via: &str, results: &[PushOutcome]) -> sqlx::Result<()> {
359	let now = now_ms();
360	let mut tx = super::begin_write(pool).await?;
361	for r in results {
362	sqlx::query("INSERT INTO pushes (repo_id, at_ms, reference, ok, reason, old_state, new_state, heads, changes, account_id, pusher, via) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)")
363	.bind(id)
364	.bind(now)
365	.bind(&r.reference)
366	.bind(r.ok)
367	.bind(&r.reason)
368	.bind(&r.old_state)
369	.bind(&r.new_state)
370	.bind(r.heads)
371	.bind(r.changes)
372	.bind(pusher.map(\|p\| p.0).filter(\|id\| *id > 0))
373	.bind(pusher.map(\|p\| p.1).unwrap_or(""))
374	.bind(via)
375	.execute(&mut *tx)
376	.await?;
377	}
378	if results.iter().any(\|r\| r.ok) {
379	sqlx::query("UPDATE repositories SET updated_ms = $1 WHERE id = $2").bind(now).bind(id).execute(&mut *tx).await?;
380	}
381	tx.commit().await
382	}
383
384	pub async fn pushes(pool: &Pool, id: i64, limit: i64) -> sqlx::Result<Vec<PushRow>> {
385	sqlx::query_as("SELECT id, at_ms, reference, ok, reason, old_state, new_state, heads, changes, pusher FROM pushes WHERE repo_id = $1 ORDER BY at_ms DESC, id DESC LIMIT $2")
386	.bind(id)
387	.bind(limit)
388	.fetch_all(pool)
389	.await
390	}
391
392	/// Topics from user input: comma or space separated, lower-cased,
393	/// letters, digits and dashes, at most 35 characters each and 20 in all.
394	/// Returns the stored form or what is wrong.
395	pub fn parse_topics(input: &str) -> Result<String, String> {
396	let mut out: Vec<String> = Vec::new();
397	for t in input.split([',', ' ', '\n', '\t']).map(\|t\| t.trim().to_ascii_lowercase()).filter(\|t\| !t.is_empty()) {
398	if t.len() > 35 \|\| !t.chars().all(\|c\| c.is_ascii_alphanumeric() \|\| c == '-') \|\| t.starts_with('-') {
399	return Err(format!("\"{t}\" is not a topic: use letters, digits and dashes, up to 35 characters"));
400	}
401	if !out.contains(&t) {
402	out.push(t);
403	}
404	}
405	if out.len() > 20 {
406	return Err("At most 20 topics.".into());
407	}
408	Ok(out.join(" "))
409	}
410
411	pub async fn set_topics(pool: &Pool, id: i64, topics: &str) -> sqlx::Result<()> {
412	sqlx::query("UPDATE repositories SET topics = $1 WHERE id = $2").bind(topics).bind(id).execute(pool).await?;
413	Ok(())
414	}
415
416	/// Star or unstar; returns whether it is now starred.
417	pub async fn toggle_star(pool: &Pool, account_id: i64, repo_id: i64) -> sqlx::Result<bool> {
418	let gone = sqlx::query("DELETE FROM stars WHERE account_id = $1 AND repo_id = $2").bind(account_id).bind(repo_id).execute(pool).await?;
419	if gone.rows_affected() > 0 {
420	return Ok(false);
421	}
422	sqlx::query("INSERT INTO stars (account_id, repo_id, at_ms) VALUES ($1, $2, $3)")
423	.bind(account_id)
424	.bind(repo_id)
425	.bind(now_ms())
426	.execute(pool)
427	.await?;
428	Ok(true)
429	}
430
431	pub async fn starred(pool: &Pool, account_id: i64, repo_id: i64) -> sqlx::Result<bool> {
432	let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stars WHERE account_id = $1 AND repo_id = $2")
433	.bind(account_id)
434	.bind(repo_id)
435	.fetch_one(pool)
436	.await?;
437	Ok(n > 0)
438	}
439
440	/// Stars and watchers of one repository.
441	pub async fn counts(pool: &Pool, repo_id: i64) -> sqlx::Result<(i64, i64)> {
442	let stars: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stars WHERE repo_id = $1").bind(repo_id).fetch_one(pool).await?;
443	let watchers = super::notify::watchers(pool, repo_id).await?.len() as i64;
444	Ok((stars, watchers))
445	}
446
447	/// Star counts of every starred repository.
448	pub async fn star_counts(pool: &Pool) -> sqlx::Result<std::collections::HashMap<i64, i64>> {
449	let rows: Vec<(i64, i64)> = sqlx::query_as("SELECT repo_id, COUNT(*) FROM stars GROUP BY repo_id").fetch_all(pool).await?;
450	Ok(rows.into_iter().collect())
451	}
452
453	/// Record that `id` is a fork of `from`.
454	pub async fn set_forked_from(pool: &Pool, id: i64, from: i64) -> sqlx::Result<()> {
455	sqlx::query("UPDATE repositories SET forked_from = $1 WHERE id = $2").bind(from).bind(id).execute(pool).await?;
456	Ok(())
457	}
458
459	/// The forks of `id`, newest activity first.
460	pub async fn forks_of(pool: &Pool, id: i64) -> sqlx::Result<Vec<RepoRow>> {
461	sqlx::query_as("SELECT * FROM repositories WHERE forked_from = $1 AND kind = 'repo' ORDER BY updated_ms DESC")
462	.bind(id)
463	.fetch_all(pool)
464	.await
465	}
466
467	/// `on`, `off`, or `` (follow hub.toml's `[flow] repos`).
468	pub async fn set_flow(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
469	sqlx::query("UPDATE repositories SET flow = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
470	Ok(())
471	}
472
473	pub async fn set_flow_priority(pool: &Pool, id: i64, priority: &str) -> sqlx::Result<()> {
474	sqlx::query("UPDATE repositories SET flow_priority = $1 WHERE id = $2").bind(priority).bind(id).execute(pool).await?;
475	Ok(())
476	}
477
478	/// Whether agents' tokens may decide the repository's approval gates (#258).
479	pub async fn set_agents_decide_gates(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
480	sqlx::query("UPDATE repositories SET agents_decide_gates = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
481	Ok(())
482	}
483
484	/// Whether a new run leaves the older runs it supersedes that are already
485	/// running to finish (#267), cancelling only the queued ones.
486	pub async fn set_flow_reuse(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
487	sqlx::query("UPDATE repositories SET flow_reuse = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
488	Ok(())
489	}
490
491	pub async fn set_flow_skip(pool: &Pool, id: i64, skip: &str) -> sqlx::Result<()> {
492	sqlx::query("UPDATE repositories SET flow_skip = $1 WHERE id = $2").bind(skip).bind(id).execute(pool).await?;
493	Ok(())
494	}
495
496	pub async fn set_flow_keep_running(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
497	sqlx::query("UPDATE repositories SET flow_keep_running = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
498	Ok(())
499	}
500
501	pub async fn set_backup(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
502	sqlx::query("UPDATE repositories SET backup = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
503	Ok(())
504	}
505
506	pub async fn set_diffusion_land(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
507	sqlx::query("UPDATE repositories SET diffusion_land = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
508	Ok(())
509	}
510
511	pub async fn set_diffusion(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {
512	sqlx::query("UPDATE repositories SET diffusion = $1 WHERE id = $2").bind(setting).bind(id).execute(pool).await?;
513	Ok(())
514	}
515
516	pub async fn set_git_mirror(pool: &Pool, id: i64, on: bool) -> sqlx::Result<()> {
517	sqlx::query("UPDATE repositories SET git_mirror = $1 WHERE id = $2").bind(on).bind(id).execute(pool).await?;
518	Ok(())
519	}
520
521	pub async fn set_git_mirror_state(pool: &Pool, id: i64, state: &str) -> sqlx::Result<()> {
522	sqlx::query("UPDATE repositories SET git_mirror_state = $1 WHERE id = $2").bind(state).bind(id).execute(pool).await?;
523	Ok(())
524	}
525
526	#[cfg(test)]
527	mod topic_tests {
528	use super::parse_topics;
529
530	#[test]
531	fn topics_are_normalised_and_checked() {
532	assert_eq!(parse_topics("Rust, vcs retro-computing,rust").unwrap(), "rust vcs retro-computing");
533	assert_eq!(parse_topics("").unwrap(), "");
534	assert!(parse_topics("c++").is_err());
535	assert!(parse_topics("-lead").is_err());
536	assert!(parse_topics(&"x".repeat(36)).is_err());
537	assert!(parse_topics(&(0..21).map(\|i\| format!("t{i}")).collect::<Vec<_>>().join(" ")).is_err());
538	}
539	}

Keyboard