Omar Sobh
637e1bdd69
Sidebar: - Each connection header now has three inline icon buttons: Sync now (spins while in flight), Edit (opens the modal), Remove (opens an inline confirm strip). Removes cascade repos via ON DELETE CASCADE. - The connection's last_sync_error surfaces as a red inline banner under the header — no more 'error status with nowhere to see why'. - Sync is POST /api/repos/connections/:id/sync (already existed); after either sync or delete the sidebar re-fetches so state stays consistent. Edit modal (RepoConnectionEditModal): - Loads GET /api/repos/connections/:id, pre-fills owner/base_url/label - PATCHes only the fields that actually changed; empty string on a Some(&str) field sends explicit null so the backend clears it - Sync-now + Remove reachable from inside the modal too - Rotating the token is out of scope: the modal says as much and points the user at delete + re-create through the wizard (the broker doesn't expose an update path, and rotating in place would require duplicating the whole broker->store_secret flow here) Backend: - GET /api/repos/connections/:id — same ConnectionSummary shape - PATCH /api/repos/connections/:id — owner/base_url use Option<Option<T>> double-nesting so 'omit = leave alone' and 'null = clear' round-trip distinctly through serde - repo_connections::update with COALESCE-per-field so the SQL matches the double-Option semantics without an OR-chain per field
730 lines
24 KiB
Rust
730 lines
24 KiB
Rust
//! 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))
|
|
}
|
|
|
|
/// `GET /api/repos/connections/:id` — full detail for the edit modal.
|
|
pub async fn get_connection(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<Json<ConnectionSummary>, ApiError> {
|
|
let c = cm_db::repo::repo_connections::get(&state.pool, id, user.workspace_id).await?;
|
|
Ok(Json(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(),
|
|
}))
|
|
}
|
|
|
|
/// `PATCH /api/repos/connections/:id` — edit owner / base_url / label on an
|
|
/// existing connection. Any field omitted from the body is left as-is;
|
|
/// explicit `null` on `owner` or `base_url` clears the value. Rotating the
|
|
/// PAT is out-of-band: delete + re-create through the wizard.
|
|
///
|
|
/// The response is the freshly-loaded connection so the client can react to
|
|
/// derived fields (`status`, `last_sync_error` cleared by a preceding sync).
|
|
pub async fn update_connection(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
Json(body): Json<UpdateConnectionRequest>,
|
|
) -> Result<Json<ConnectionSummary>, ApiError> {
|
|
let owner = body
|
|
.owner
|
|
.map(|opt| opt.map(|s| s.trim().to_string()).filter(|s| !s.is_empty()));
|
|
let base_url = body
|
|
.base_url
|
|
.map(|opt| opt.map(|s| s.trim().to_string()).filter(|s| !s.is_empty()));
|
|
let label = body
|
|
.label
|
|
.as_deref()
|
|
.map(str::trim)
|
|
.filter(|s| !s.is_empty());
|
|
let owner_ref = owner.as_ref().map(|opt| opt.as_deref());
|
|
let base_url_ref = base_url.as_ref().map(|opt| opt.as_deref());
|
|
let ok = cm_db::repo::repo_connections::update(
|
|
&state.pool,
|
|
id,
|
|
user.workspace_id,
|
|
owner_ref,
|
|
base_url_ref,
|
|
label,
|
|
)
|
|
.await?;
|
|
if !ok {
|
|
return Err(ApiError::NotFound);
|
|
}
|
|
get_connection(State(state), Authed(user), Path(id)).await
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct UpdateConnectionRequest {
|
|
/// `Some(None)` clears; `None` leaves unchanged.
|
|
#[serde(default, deserialize_with = "de_double_option")]
|
|
pub owner: Option<Option<String>>,
|
|
#[serde(default, deserialize_with = "de_double_option")]
|
|
pub base_url: Option<Option<String>>,
|
|
#[serde(default)]
|
|
pub label: Option<String>,
|
|
}
|
|
|
|
// serde default treats a missing field as `None` and an explicit `null` as
|
|
// `Some(None)` when the target type is Option<Option<T>>. Manual deserializer
|
|
// is needed because serde otherwise conflates the two.
|
|
fn de_double_option<'de, D>(d: D) -> Result<Option<Option<String>>, D::Error>
|
|
where
|
|
D: serde::Deserializer<'de>,
|
|
{
|
|
Option::<Option<String>>::deserialize(d)
|
|
}
|
|
|
|
/// `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() {
|
|
"gitea" => sync_gitea(state, workspace_id, &conn, secret_ref).await,
|
|
"github" => sync_github(state, workspace_id, &conn, secret_ref).await,
|
|
other => Err(format!("provider '{other}' not yet supported for sync")),
|
|
}
|
|
}
|
|
|
|
/// Gitea sync — Gitea's REST v1 is close enough to GitHub's that we share the
|
|
/// upsert shape but diverge on URL construction and a couple of field names
|
|
/// (`stars_count`, `forks_count` vs GitHub's `stargazers_count`, `forks_count`;
|
|
/// `full_name` present on every row; `owner.login` matches). Base URL must
|
|
/// point at the instance root — we append `/api/v1` ourselves so callers
|
|
/// don't have to remember which providers include the API prefix.
|
|
///
|
|
/// Fleet default is the redclaw Gitea (`git.redclaw.dev`). Gitea gets first-
|
|
/// class treatment because it's what most of the workspace's repos live on.
|
|
async fn sync_gitea(
|
|
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}"))?;
|
|
|
|
let base_root = conn
|
|
.base_url
|
|
.as_deref()
|
|
.map(|s| s.trim().trim_end_matches('/'))
|
|
.filter(|s| !s.is_empty())
|
|
.ok_or_else(|| "gitea requires a base_url".to_string())?;
|
|
// Accept either `https://git.example.com` or `https://git.example.com/api/v1`.
|
|
let api_base = if base_root.ends_with("/api/v1") {
|
|
base_root.to_string()
|
|
} else {
|
|
format!("{base_root}/api/v1")
|
|
};
|
|
|
|
let mut page = 1u32;
|
|
let per_page = 50u32;
|
|
let mut upserted = 0usize;
|
|
loop {
|
|
let url = match conn.owner.as_deref() {
|
|
Some(owner) => format!("{api_base}/orgs/{owner}/repos?limit={per_page}&page={page}"),
|
|
None => format!("{api_base}/repos/search?limit={per_page}&page={page}"),
|
|
};
|
|
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!("Gitea {status}: {body}"));
|
|
}
|
|
// `/orgs/:owner/repos` returns an array; `/repos/search` wraps it in
|
|
// `{data: [...]}` with an `ok` flag.
|
|
let arr = if let Some(a) = body.as_array() {
|
|
a.clone()
|
|
} else if let Some(a) = body.get("data").and_then(|v| v.as_array()) {
|
|
a.clone()
|
|
} else {
|
|
break;
|
|
};
|
|
if arr.is_empty() {
|
|
break;
|
|
}
|
|
let n = arr.len();
|
|
for repo in &arr {
|
|
if let Err(e) =
|
|
upsert_gitea_repo(&state.pool, workspace_id, conn.id, &conn.provider, repo).await
|
|
{
|
|
eprintln!(
|
|
"repos: gitea 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;
|
|
// Bounded first-sync latency: cap at 20 pages (~1000 repos on Gitea).
|
|
if page > 20 {
|
|
break;
|
|
}
|
|
}
|
|
Ok(upserted)
|
|
}
|
|
|
|
async fn upsert_gitea_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").or_else(|| v.get("username")))
|
|
.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())
|
|
.filter(|s| !s.is_empty())
|
|
.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);
|
|
// Gitea uses `stars_count` (not `stargazers_count`) but we accept either
|
|
// for forward-compat across versions.
|
|
let stars = repo
|
|
.get("stars_count")
|
|
.or_else(|| 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(())
|
|
}
|
|
|
|
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(())
|
|
}
|