repos: backend — schema, /api/repos routes + GitHub sync provider
Migration 0034: two tables. repo_connections carries the workspace's per-provider config (owner, base_url, label, last_synced_at, last_sync_error) and points at an app_connections row for the PAT. repos is the per-connection cache with (connection_id, external_id) unique so upsert is idempotent across re-syncs. Cascading deletes clean up cleanly on connection removal. cm-secrets grows a FetchAuthorized op — GET with the stored PAT injected as bearer, returns status + JSON body without ever exposing the credential to cm-api. This is the least-privilege door for read-only provider APIs (list repos), distinct from the InvokeHttp path that still requires a single-use approval grant for outbound writes. cm-api::routes::repos wires: - POST /api/repos/connections (broker store_secret + insert both rows + initial sync + mark_synced) - GET /api/repos/connections - DELETE /api/repos/connections/:id - POST /api/repos/connections/:id/sync - GET /api/repos (500 cap, newest provider_updated first) - GET /api/repos/:id (full detail incl. clone_url + html_url) GitHub provider inline for v1 — paginated pull of /orgs/:owner/repos (when owner set) or /user/repos (when absent), 100/page, capped at 20 pages (~2k repos) to keep first-sync latency bounded. Non-2xx surface back to the caller as sync_error; parse failures are best-effort per repo (skipped, logged, don't abort the batch). Gitea + GitLab providers land in a follow-up — mostly URL swap + response-shape adapter.
This commit is contained in:
Omar Sobh
@@ -0,0 +1,458 @@
|
||||
//! Repos — workspace-scoped view of connected git providers and the
|
||||
//! repositories they own.
|
||||
//!
|
||||
//! POST /api/repos/connections create a provider connection
|
||||
//! GET /api/repos/connections list connections
|
||||
//! DELETE /api/repos/connections/:id remove a connection (cascades to repos)
|
||||
//! POST /api/repos/connections/:id/sync refetch repos from the provider
|
||||
//! GET /api/repos list every repo in the workspace
|
||||
//! GET /api/repos/:id one repo detail
|
||||
//!
|
||||
//! Provider credentials (PATs) live in the secret broker; only their ref is
|
||||
//! kept in `app_connections`. Repo fetching goes through
|
||||
//! `BrokerClient::fetch_authorized`, so the PAT never leaves the broker.
|
||||
//! v1 supports GitHub inline; Gitea + GitLab are next (mostly a URL swap +
|
||||
//! response shape adapter).
|
||||
|
||||
use axum::extract::{Path, State};
|
||||
use axum::http::StatusCode;
|
||||
use axum::Json;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{ApiError, AppState, Authed};
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct CreateConnectionRequest {
|
||||
/// `github` | `gitea` | `gitlab`
|
||||
pub provider: String,
|
||||
pub token: String,
|
||||
/// Optional org/user login. When absent GitHub pulls `/user/repos`.
|
||||
#[serde(default)]
|
||||
pub owner: Option<String>,
|
||||
/// Optional base URL for self-hosted providers.
|
||||
#[serde(default)]
|
||||
pub base_url: Option<String>,
|
||||
/// Optional human label — defaults to a "<provider>/<owner>" pattern.
|
||||
#[serde(default)]
|
||||
pub label: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct ConnectionSummary {
|
||||
pub id: String,
|
||||
pub provider: String,
|
||||
pub owner: Option<String>,
|
||||
pub base_url: Option<String>,
|
||||
pub label: String,
|
||||
pub status: String,
|
||||
pub last_synced_at: Option<String>,
|
||||
pub last_sync_error: Option<String>,
|
||||
pub created_at: String,
|
||||
}
|
||||
|
||||
fn known_provider(p: &str) -> bool {
|
||||
matches!(p, "github" | "gitea" | "gitlab")
|
||||
}
|
||||
|
||||
fn default_label(provider: &str, owner: Option<&str>) -> String {
|
||||
match owner {
|
||||
Some(o) => format!("{provider}/{o}"),
|
||||
None => provider.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /api/repos/connections` — store the PAT in the broker, create the
|
||||
/// app_connections + repo_connections rows, and kick off an initial sync.
|
||||
pub async fn create_connection(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Json(body): Json<CreateConnectionRequest>,
|
||||
) -> Result<(StatusCode, Json<Value>), ApiError> {
|
||||
let provider = body.provider.trim().to_lowercase();
|
||||
if !known_provider(&provider) {
|
||||
return Err(ApiError::BadRequest);
|
||||
}
|
||||
if body.token.trim().is_empty() {
|
||||
return Err(ApiError::BadRequest);
|
||||
}
|
||||
let owner = body
|
||||
.owner
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty());
|
||||
let base_url = body
|
||||
.base_url
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty());
|
||||
let label = body
|
||||
.label
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_owned)
|
||||
.unwrap_or_else(|| default_label(&provider, owner));
|
||||
|
||||
// Stash the PAT in the broker; only the ref persists in Postgres.
|
||||
let socket = state.broker_socket.as_ref().ok_or(ApiError::Internal)?;
|
||||
let mut broker = cm_secrets::BrokerClient::connect(socket)
|
||||
.await
|
||||
.map_err(|_| ApiError::Internal)?;
|
||||
let secret_ref = broker
|
||||
.store_secret(
|
||||
user.workspace_id,
|
||||
&format!("{provider}_pat"),
|
||||
body.token.trim(),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| ApiError::Internal)?;
|
||||
|
||||
let app_conn = cm_db::repo::connections::insert(
|
||||
&state.pool,
|
||||
user.workspace_id,
|
||||
None,
|
||||
&provider,
|
||||
"keys",
|
||||
secret_ref,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let repo_conn_id = cm_db::repo::repo_connections::insert(
|
||||
&state.pool,
|
||||
cm_db::repo::repo_connections::NewRepoConnection {
|
||||
workspace_id: user.workspace_id,
|
||||
app_connection_id: app_conn.id,
|
||||
provider: &provider,
|
||||
owner,
|
||||
base_url,
|
||||
label: &label,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
// First sync — best-effort. Errors mark the connection with the message;
|
||||
// the caller can retry via POST /:id/sync.
|
||||
let sync_result = sync_connection(&state, user.workspace_id, repo_conn_id, secret_ref).await;
|
||||
let (synced, sync_error) = match sync_result {
|
||||
Ok(n) => (n, None),
|
||||
Err(e) => (0, Some(e)),
|
||||
};
|
||||
let _ = cm_db::repo::repo_connections::mark_synced(
|
||||
&state.pool,
|
||||
repo_conn_id,
|
||||
sync_error.as_deref(),
|
||||
)
|
||||
.await;
|
||||
|
||||
Ok((
|
||||
StatusCode::CREATED,
|
||||
Json(serde_json::json!({
|
||||
"id": repo_conn_id.to_string(),
|
||||
"provider": provider,
|
||||
"owner": owner,
|
||||
"synced": synced,
|
||||
"sync_error": sync_error,
|
||||
})),
|
||||
))
|
||||
}
|
||||
|
||||
/// `GET /api/repos/connections` — list every provider connection in the workspace.
|
||||
pub async fn list_connections(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
) -> Result<Json<Vec<ConnectionSummary>>, ApiError> {
|
||||
let rows =
|
||||
cm_db::repo::repo_connections::list_for_workspace(&state.pool, user.workspace_id).await?;
|
||||
let out = rows
|
||||
.into_iter()
|
||||
.map(|c| ConnectionSummary {
|
||||
id: c.id.to_string(),
|
||||
provider: c.provider,
|
||||
owner: c.owner,
|
||||
base_url: c.base_url,
|
||||
label: c.label,
|
||||
status: match c.last_sync_error.as_deref() {
|
||||
Some(_) => "error".into(),
|
||||
None if c.last_synced_at.is_some() => "connected".into(),
|
||||
None => "pending".into(),
|
||||
},
|
||||
last_synced_at: c.last_synced_at.and_then(|t| t.format(&Rfc3339).ok()),
|
||||
last_sync_error: c.last_sync_error,
|
||||
created_at: c.created_at.format(&Rfc3339).unwrap_or_default(),
|
||||
})
|
||||
.collect();
|
||||
Ok(Json(out))
|
||||
}
|
||||
|
||||
/// `DELETE /api/repos/connections/:id` — remove the repo_connections row.
|
||||
/// `ON DELETE CASCADE` cleans out its repos; the underlying app_connections
|
||||
/// row + broker secret stay (the workspace may reuse the token elsewhere).
|
||||
pub async fn delete_connection(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(id): Path<Uuid>,
|
||||
) -> Result<StatusCode, ApiError> {
|
||||
let removed = cm_db::repo::repo_connections::delete(&state.pool, id, user.workspace_id).await?;
|
||||
if removed {
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
} else {
|
||||
Err(ApiError::NotFound)
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /api/repos/connections/:id/sync` — refetch from the provider.
|
||||
pub async fn sync_now(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(id): Path<Uuid>,
|
||||
) -> Result<Json<Value>, ApiError> {
|
||||
let conn = cm_db::repo::repo_connections::get(&state.pool, id, user.workspace_id).await?;
|
||||
let app_conn =
|
||||
cm_db::repo::connections::get(&state.pool, conn.app_connection_id, user.workspace_id)
|
||||
.await?;
|
||||
let secret_ref = app_conn.secret_ref.ok_or(ApiError::Conflict)?;
|
||||
let result = sync_connection(&state, user.workspace_id, id, secret_ref).await;
|
||||
let (synced, sync_error) = match result {
|
||||
Ok(n) => (n, None),
|
||||
Err(e) => (0, Some(e)),
|
||||
};
|
||||
let _ =
|
||||
cm_db::repo::repo_connections::mark_synced(&state.pool, id, sync_error.as_deref()).await;
|
||||
Ok(Json(serde_json::json!({
|
||||
"synced": synced,
|
||||
"sync_error": sync_error,
|
||||
})))
|
||||
}
|
||||
|
||||
/// `GET /api/repos` — every cached repo in the workspace, newest updated first.
|
||||
pub async fn list_repos(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
) -> Result<Json<Value>, ApiError> {
|
||||
let rows = cm_db::repo::repos::list_for_workspace(&state.pool, user.workspace_id, 500).await?;
|
||||
let out: Vec<Value> = rows
|
||||
.into_iter()
|
||||
.map(|r| {
|
||||
serde_json::json!({
|
||||
"id": r.id.to_string(),
|
||||
"connection_id": r.connection_id.to_string(),
|
||||
"provider": r.provider,
|
||||
"owner": r.owner,
|
||||
"name": r.name,
|
||||
"private": r.private,
|
||||
"description": r.description,
|
||||
"default_branch": r.default_branch,
|
||||
"stars": r.stars,
|
||||
"forks": r.forks,
|
||||
"updated_at": r.provider_updated_at.and_then(|t| t.format(&Rfc3339).ok()),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
Ok(Json(Value::Array(out)))
|
||||
}
|
||||
|
||||
/// `GET /api/repos/:id` — one repo detail (includes clone_url + html_url).
|
||||
pub async fn get_repo(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(id): Path<Uuid>,
|
||||
) -> Result<Json<Value>, ApiError> {
|
||||
let r = cm_db::repo::repos::get(&state.pool, id, user.workspace_id).await?;
|
||||
Ok(Json(serde_json::json!({
|
||||
"id": r.id.to_string(),
|
||||
"connection_id": r.connection_id.to_string(),
|
||||
"provider": r.provider,
|
||||
"owner": r.owner,
|
||||
"name": r.name,
|
||||
"private": r.private,
|
||||
"description": r.description,
|
||||
"default_branch": r.default_branch,
|
||||
"clone_url": r.clone_url,
|
||||
"html_url": r.html_url,
|
||||
"stars": r.stars,
|
||||
"forks": r.forks,
|
||||
"updated_at": r.provider_updated_at.and_then(|t| t.format(&Rfc3339).ok()),
|
||||
"last_synced_at": r.last_synced_at.format(&Rfc3339).ok(),
|
||||
})))
|
||||
}
|
||||
|
||||
// ── provider sync ────────────────────────────────────────────────────────
|
||||
|
||||
/// Fetch repos from the provider (via the broker) + upsert into the cache.
|
||||
/// Returns the number of upserted rows. Errors on network / provider errors —
|
||||
/// the caller stashes the message on repo_connections.last_sync_error.
|
||||
async fn sync_connection(
|
||||
state: &AppState,
|
||||
workspace_id: cm_domain::WorkspaceId,
|
||||
repo_conn_id: Uuid,
|
||||
secret_ref: Uuid,
|
||||
) -> Result<usize, String> {
|
||||
let conn = cm_db::repo::repo_connections::get(&state.pool, repo_conn_id, workspace_id)
|
||||
.await
|
||||
.map_err(|e| format!("load connection: {e}"))?;
|
||||
match conn.provider.as_str() {
|
||||
"github" => sync_github(state, workspace_id, &conn, secret_ref).await,
|
||||
other => Err(format!("provider '{other}' not yet supported for sync")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn sync_github(
|
||||
state: &AppState,
|
||||
workspace_id: cm_domain::WorkspaceId,
|
||||
conn: &cm_db::repo::repo_connections::RepoConnection,
|
||||
secret_ref: Uuid,
|
||||
) -> Result<usize, String> {
|
||||
let socket = state
|
||||
.broker_socket
|
||||
.as_ref()
|
||||
.ok_or_else(|| "broker socket unavailable".to_string())?;
|
||||
let mut broker = cm_secrets::BrokerClient::connect(socket)
|
||||
.await
|
||||
.map_err(|e| format!("broker: {e}"))?;
|
||||
|
||||
// GitHub API paginates with `per_page` + `page`. 100 is the max; stop
|
||||
// when we get a short page (< per_page) — sufficient for v1.
|
||||
let base = conn.base_url.as_deref().unwrap_or("https://api.github.com");
|
||||
let mut page = 1u32;
|
||||
let per_page = 100u32;
|
||||
let mut upserted = 0usize;
|
||||
loop {
|
||||
let url = match conn.owner.as_deref() {
|
||||
Some(owner) => format!(
|
||||
"{base}/orgs/{owner}/repos?per_page={per_page}&page={page}&type=all"
|
||||
),
|
||||
None => format!("{base}/user/repos?per_page={per_page}&page={page}&affiliation=owner,organization_member"),
|
||||
};
|
||||
let (status, body) = broker
|
||||
.fetch_authorized(secret_ref, &url)
|
||||
.await
|
||||
.map_err(|e| format!("broker fetch: {e}"))?;
|
||||
if status == 404 && conn.owner.is_some() {
|
||||
return Err(format!(
|
||||
"org '{}' not found or PAT lacks access",
|
||||
conn.owner.as_deref().unwrap_or("")
|
||||
));
|
||||
}
|
||||
if !(200..300).contains(&status) {
|
||||
return Err(format!("GitHub {status}: {body}"));
|
||||
}
|
||||
let arr = match body.as_array() {
|
||||
Some(a) => a.clone(),
|
||||
None => break,
|
||||
};
|
||||
if arr.is_empty() {
|
||||
break;
|
||||
}
|
||||
let n = arr.len();
|
||||
for repo in &arr {
|
||||
if let Err(e) =
|
||||
upsert_github_repo(&state.pool, workspace_id, conn.id, &conn.provider, repo).await
|
||||
{
|
||||
eprintln!(
|
||||
"repos: upsert failed for {}: {e}",
|
||||
repo.get("full_name")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("?")
|
||||
);
|
||||
continue;
|
||||
}
|
||||
upserted += 1;
|
||||
}
|
||||
if (n as u32) < per_page {
|
||||
break;
|
||||
}
|
||||
page += 1;
|
||||
// Cap at 20 pages (~2k repos) to keep first-sync latency bounded.
|
||||
if page > 20 {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(upserted)
|
||||
}
|
||||
|
||||
async fn upsert_github_repo(
|
||||
pool: &sqlx::PgPool,
|
||||
workspace_id: cm_domain::WorkspaceId,
|
||||
connection_id: Uuid,
|
||||
provider: &str,
|
||||
repo: &Value,
|
||||
) -> Result<(), cm_db::DbError> {
|
||||
let external_id = repo
|
||||
.get("id")
|
||||
.and_then(|v| v.as_i64())
|
||||
.map(|n| n.to_string())
|
||||
.unwrap_or_default();
|
||||
let owner = repo
|
||||
.get("owner")
|
||||
.and_then(|v| v.get("login"))
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
let name = repo
|
||||
.get("name")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
if external_id.is_empty() || name.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let description = repo
|
||||
.get("description")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(str::to_owned);
|
||||
let default_branch = repo
|
||||
.get("default_branch")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(str::to_owned);
|
||||
let clone_url = repo
|
||||
.get("clone_url")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(str::to_owned);
|
||||
let html_url = repo
|
||||
.get("html_url")
|
||||
.and_then(|v| v.as_str())
|
||||
.map(str::to_owned);
|
||||
let private = repo
|
||||
.get("private")
|
||||
.and_then(|v| v.as_bool())
|
||||
.unwrap_or(false);
|
||||
let stars = repo
|
||||
.get("stargazers_count")
|
||||
.and_then(|v| v.as_i64())
|
||||
.unwrap_or(0) as i32;
|
||||
let forks = repo
|
||||
.get("forks_count")
|
||||
.and_then(|v| v.as_i64())
|
||||
.unwrap_or(0) as i32;
|
||||
let provider_updated_at = repo
|
||||
.get("updated_at")
|
||||
.and_then(|v| v.as_str())
|
||||
.and_then(|s| {
|
||||
time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339).ok()
|
||||
});
|
||||
|
||||
cm_db::repo::repos::upsert(
|
||||
pool,
|
||||
cm_db::repo::repos::RepoUpsert {
|
||||
connection_id,
|
||||
workspace_id,
|
||||
provider,
|
||||
external_id: &external_id,
|
||||
owner: &owner,
|
||||
name: &name,
|
||||
description: description.as_deref(),
|
||||
default_branch: default_branch.as_deref(),
|
||||
clone_url: clone_url.as_deref(),
|
||||
html_url: html_url.as_deref(),
|
||||
private,
|
||||
stars,
|
||||
forks,
|
||||
provider_updated_at,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user