use std::path::Path;
use std::sync::Arc;
use anyhow::Context as _;
use axum::extract::State;
use axum::http::StatusCode;
use axum::Json;
use super::{bad as fail, Cur, Reply, Shared};
use crate::api;
use crate::config::RepoEntry;
use crate::events::{Event, EventSink, Kind, Level};
use crate::jobs::Status;
use crate::worktree;
fn absolute(p: &str) -> String {
let p = crate::config::expand_tilde(p.trim());
std::fs::canonicalize(&p)
.map(|c| c.to_string_lossy().into_owned())
.unwrap_or(p)
}
pub(super) async fn list(State(app): State<Shared>, Cur(s): Cur) -> Json<api::Repos> {
let repos = app.repos();
let indexed: std::collections::HashMap<String, usize> =
s.data.graph.repos().into_iter().collect();
let meta = match super::graph_uri(&s) {
Some(u) => crate::index::read_indexed_repos(&u)
.await
.unwrap_or_default(),
None => Vec::new(),
};
let commits: std::collections::HashMap<String, String> = meta
.iter()
.map(|r| (r.name.clone(), r.commit.clone()))
.collect();
let branches: std::collections::HashMap<String, String> = meta
.iter()
.filter_map(|r| Some((r.name.clone(), r.branch.clone()?)))
.collect();
let symbols = indexed.clone();
let listed = repos.clone();
let rows = tokio::task::spawn_blocking(move || {
listed
.iter()
.map(|r| {
let p = Path::new(&r.path);
let name = worktree::name(p);
api::RepoRow {
path: r.path.clone(),
exists: p.join(".git").exists(),
branch: worktree::default_branch(p)
.ok()
.or_else(|| branches.get(&name).cloned()),
managed: r.remote.is_some(),
cloned: p.exists(),
your_branch: worktree::current_branch(p),
remote: r
.remote
.clone()
.or_else(|| worktree::remote_url(p))
.map(|u| crate::storage::without_userinfo(&u)),
symbols: symbols.get(&name).copied(),
indexed_commit: commits.get(&name).cloned(),
name,
}
})
.collect::<Vec<_>>()
})
.await
.unwrap_or_default();
let listed: Vec<String> = rows.iter().map(|r| r.name.clone()).collect();
let mine: Vec<String> = repos
.iter()
.filter_map(|r| {
r.remote
.clone()
.or_else(|| worktree::remote_url(Path::new(&r.path)))
})
.map(|u| same_repo(&u))
.collect();
let team_missing: Vec<String> = app
.team()
.repos
.into_iter()
.filter(|u| !mine.contains(&same_repo(u)))
.collect();
Json(api::Repos {
repos: rows,
unlisted: unlisted(&indexed, &meta, &listed),
team_missing,
root: s.data.root.clone(),
})
}
fn unlisted(
indexed: &std::collections::HashMap<String, usize>,
meta: &[crate::index::IndexedRepo],
listed: &[String],
) -> Vec<api::UnlistedRepo> {
let mut names: Vec<&String> = indexed.keys().collect();
names.sort();
names
.into_iter()
.filter(|n| !listed.contains(n))
.map(|n| {
let m = meta.iter().find(|r| &r.name == n);
api::UnlistedRepo {
name: n.clone(),
symbols: indexed.get(n).copied(),
commit: m.map(|r| r.commit.clone()),
remote: m
.and_then(|r| r.remote.as_deref())
.map(crate::storage::without_userinfo),
}
})
.collect()
}
pub(super) async fn scan(
State(app): State<Shared>,
Json(q): Json<api::ScanRequest>,
) -> Reply<api::Scan> {
let dir = absolute(&q.dir);
if !Path::new(&dir).is_dir() {
return Err(fail(anyhow::anyhow!("{dir} is not a folder")));
}
let listed: Vec<String> = app.repos().iter().map(|r| absolute(&r.path)).collect();
let found = tokio::task::spawn_blocking(move || {
worktree::scan(Path::new(&dir), q.depth.unwrap_or(2).clamp(1, 4))
.into_iter()
.map(|p| {
let path = p.to_string_lossy().into_owned();
api::ScanHit {
name: worktree::name(&p),
listed: listed.contains(&path),
remote: worktree::remote_url(&p).map(|u| crate::storage::without_userinfo(&u)),
path,
}
})
.collect::<Vec<_>>()
})
.await
.unwrap_or_default();
Ok(Json(api::Scan { ok: true, found }))
}
pub(super) async fn change(
who: super::Member,
State(app): State<Shared>,
Json(c): Json<api::RepoChange>,
) -> Reply<api::Done> {
let mut list = app.repos();
let by = who.as_ref().map(|w| w.email.clone());
let named: Vec<String> = c.add.iter().map(|p| worktree::name(Path::new(p))).collect();
let teams = c.teams.clone();
let shared = app.clone();
let run = tokio::task::spawn_blocking(move || {
for p in &c.add {
let path = absolute(p);
anyhow::ensure!(
Path::new(&path).join(".git").exists(),
"{path} is not a git repository"
);
if !list.iter().any(|r| absolute(&r.path) == path) {
list.push(RepoEntry { path, remote: None });
}
}
let gone: Vec<String> = c.remove.iter().map(|p| absolute(p)).collect();
list.retain(|r| {
let drop = gone.contains(&absolute(&r.path));
if drop {
let _ = worktree::remove(&crate::config::cache_dir(), Path::new(&r.path));
}
!drop
});
app.set_repos(by.as_deref(), &list)?;
anyhow::Ok(list.len())
})
.await;
let run = match run.map_err(anyhow::Error::from).and_then(|r| r) {
Ok(n) => label_added(&shared, &who, &named, &teams).await.map(|_| n),
Err(e) => Err(e),
};
match run {
Ok(n) => {
let what = if n == 1 { "repository" } else { "repositories" };
Ok(Json(api::Done::new(format!(
"saved; {n} {what} in the list"
))))
}
Err(e) => Err(fail(e)),
}
}
pub(super) async fn cancel(State(app): State<Shared>) -> Json<api::Done> {
Json(api::Done::new(match app.jobs.running() {
Some(job) => {
job.cancel.cancel();
"cancelling; work already saved stays"
}
None => "nothing is indexing",
}))
}
pub(super) async fn job(State(app): State<Shared>) -> Json<api::IndexState> {
let queued: Vec<api::QueuedJob> = app
.jobs
.queued()
.iter()
.enumerate()
.map(|(i, q)| api::QueuedJob {
id: q.id.clone(),
kind: q.kind.clone(),
position: i + 1,
started_by: q.started_by.clone(),
created: q.created.clone(),
})
.collect();
let Some((j, events)) = app.jobs.latest() else {
return Json(api::IndexState::Idle {
running: false,
events: Vec::new(),
queued,
});
};
Json(api::IndexState::Latest(api::LatestJob {
id: j.id,
running: !j.status.finished(),
cancelled: j.status == Status::Cancelled,
ok: j.status.finished().then_some(j.status == Status::Succeeded),
status: j.status,
events,
started: j.started.unwrap_or(j.created),
finished: j.finished,
queued,
}))
}
pub(super) async fn start(
who: super::Member,
rid: Option<axum::Extension<super::RequestId>>,
State(app): State<Shared>,
Json(q): Json<api::IndexRequest>,
) -> Reply<api::IndexStarted> {
let by = who.map(|w| w.email.clone());
let rid = rid.map(|r| r.0 .0);
match begin(
&app,
q.llm,
q.drop_unlisted,
q.rewrite,
q.paths,
q.steps,
None,
by,
rid,
)
.await
{
Ok(s) => {
let message = match (s.position, s.coalesced) {
(None, _) => "indexing started".to_string(),
(Some(n), false) => format!("queued (position {n})"),
(Some(n), true) => format!("the same indexing is already queued (position {n})"),
};
let r = s.job.record();
Ok(Json(api::IndexStarted {
ok: true,
message,
id: r.id,
status: r.status,
position: s.position,
}))
}
Err(e) => Err(fail(e)),
}
}
#[allow(clippy::too_many_arguments)]
pub(super) async fn begin(
app: &Shared,
llm: bool,
drop_unlisted: bool,
rewrite: Vec<crate::prompts::Task>,
paths: Vec<String>,
steps: Vec<crate::prompts::Task>,
why: Option<String>,
by: Option<String>,
request_id: Option<String>,
) -> anyhow::Result<crate::jobs::Submitted> {
anyhow::ensure!(
llm || rewrite.is_empty(),
"rewriting needs the LLM steps; start it with them"
);
anyhow::ensure!(
paths.is_empty() || app.server.is_none(),
"a team server indexes its own repository list; add repositories on its Repositories page"
);
let session = app.current.read().await.clone();
session.data.root.clone().context("read-only UI")?;
if !drop_unlisted && paths.is_empty() {
let indexed = session.data.graph.repos().into_iter().collect();
let listed: Vec<String> = app
.repos()
.iter()
.map(|r| worktree::name(Path::new(&r.path)))
.collect();
let missing: Vec<String> = unlisted(&indexed, &[], &listed)
.into_iter()
.map(|r| r.name)
.collect();
anyhow::ensure!(
missing.is_empty(),
"no folder here for {}, which the index has; add them first, or drop them from the index",
missing.join(", ")
);
}
let key = format!(
"llm={llm} drop_unlisted={drop_unlisted} rewrite={rewrite:?} paths={paths:?} steps={steps:?}"
);
let request = serde_json::to_value(IndexRequest {
llm,
drop_unlisted,
rewrite: rewrite.clone(),
paths: paths.clone(),
steps: steps.clone(),
})?;
app.jobs.submit_resumable(
"index",
&key,
by,
request_id,
Some(request),
index_work(app.clone(), llm, rewrite, paths, steps, why),
)
}
#[derive(serde::Serialize, serde::Deserialize)]
struct IndexRequest {
llm: bool,
drop_unlisted: bool,
#[serde(default)]
rewrite: Vec<crate::prompts::Task>,
#[serde(default)]
paths: Vec<String>,
#[serde(default)]
steps: Vec<crate::prompts::Task>,
}
pub(super) async fn carry_on(app: Shared) {
loop {
tokio::time::sleep(std::time::Duration::from_secs(30)).await;
for job in app.jobs.resumable() {
if job.0.kind != "index" {
continue;
}
let Some(r) = job
.0
.request
.clone()
.and_then(|r| serde_json::from_value::<IndexRequest>(r).ok())
else {
continue;
};
let key = format!(
"llm={} drop_unlisted={} rewrite={:?} paths={:?} steps={:?}",
r.llm, r.drop_unlisted, r.rewrite, r.paths, r.steps
);
if let Err(e) = app.jobs.resume(
&key,
job,
index_work(app.clone(), r.llm, r.rewrite, r.paths, r.steps, None),
) {
crate::ui::warn(&format!("continuing a paused job: {e:#}"));
}
}
}
}
fn index_work(
app: Shared,
llm: bool,
rewrite: Vec<crate::prompts::Task>,
paths: Vec<String>,
steps: Vec<crate::prompts::Task>,
why: Option<String>,
) -> impl FnOnce(Arc<crate::jobs::Running>) -> futures::future::BoxFuture<'static, anyhow::Result<()>>
+ Send
+ 'static {
move |job| {
Box::pin(async move {
if let Some(w) = why {
job.emit(Event::new(Level::Info, Kind::Note, w));
}
let root = app
.current
.read()
.await
.data
.root
.clone()
.context("read-only UI")?;
let team = crate::team::load(&root)
.await
.map(|t| t.settings)
.unwrap_or_default();
let llm_key = llm
.then(|| crate::credentials::llm_key(team.name.as_deref(), app.server.as_deref()).0)
.flatten();
let git = app
.server
.as_ref()
.map(|s| {
s.secrets
.secret_names()
.into_iter()
.filter_map(|n| {
let host = n.strip_prefix("git:")?.to_string();
Some((host, s.secrets.secret(&n)?))
})
.collect()
})
.unwrap_or_default();
let spec = crate::indexer::Spec {
root: root.clone(),
repos: app.repos(),
llm,
llm_key,
git,
rewrite,
paths,
steps,
};
crate::indexer::run(&spec, job.clone()).await?;
if let Err(e) = crate::ledger::Ledger::open(&root)?.compact().await {
job.emit(Event::new(
Level::Warn,
Kind::Warning,
format!("ledger compaction failed: {e:#}"),
));
}
if let Err(e) = app.reload(&root).await {
job.emit(Event::new(
Level::Error,
Kind::Error,
format!("reload failed: {e:#}"),
));
}
Ok(())
})
}
}
pub(super) async fn watch(app: Shared) {
let mut last = tokio::time::Instant::now();
loop {
tokio::time::sleep(std::time::Duration::from_secs(30)).await;
let settings = app.settings();
if !settings.auto_index || last.elapsed() < settings.auto_index_every {
continue;
}
last = tokio::time::Instant::now();
let result = match moved(&app).await {
Ok(m) if m.is_empty() => "every default branch is indexed".to_string(),
Ok(m) => {
let why = format!("Automatic re-index: {} moved", m.join(", "));
match begin(
&app,
settings.auto_index_llm,
false,
Vec::new(),
Vec::new(),
Vec::new(),
Some(why.clone()),
None,
None,
)
.await
{
Ok(s) if s.coalesced => format!("{why}; a re-index is already queued"),
Ok(s) if s.position.is_some() => format!("{why}; queued"),
Ok(_) => why,
Err(e) => format!("{why}, not started: {e:#}"),
}
}
Err(e) => format!("check failed: {e:#}"),
};
*app.watch.lock().await = api::WatchResult {
checked: Some(chrono::Utc::now().to_rfc3339()),
result,
};
}
}
async fn moved(app: &Shared) -> anyhow::Result<Vec<String>> {
let uri = super::graph_uri(&app.current.read().await.clone()).context("no index")?;
let indexed = crate::index::read_repo_commits(&uri)
.await
.unwrap_or_default();
let repos = app.repos();
let creds = crate::config::git_credentials(app.server.clone());
tokio::task::spawn_blocking(move || {
let mut out = Vec::new();
for r in &repos {
let p = Path::new(&r.path);
let name = worktree::name(p);
let head = match (p.exists(), &r.remote) {
(true, _) => worktree::remote_default(p, creds.as_ref())?.1,
(false, Some(url)) => worktree::remote_head(url, creds.as_ref())?.1,
(false, None) => {
out.push(name);
continue;
}
};
if indexed
.get(&name)
.is_none_or(|c| !head.starts_with(c.as_str()) && !c.starts_with(&head))
{
out.push(name);
}
}
anyhow::Ok(out)
})
.await?
}
pub(super) async fn watch_state(State(app): State<Shared>) -> Json<api::IndexWatch> {
let settings = app.settings();
Json(api::IndexWatch {
enabled: settings.auto_index,
minutes: settings.auto_index_every.as_secs() / 60,
last: app.watch.lock().await.clone(),
})
}
fn same_repo(url: &str) -> String {
crate::hosting::managed_path(Path::new(&crate::config::data_dir()), url)
.map(|p| p.to_string_lossy().into_owned())
.unwrap_or_else(|_| url.to_string())
}
pub(super) async fn remote_list(
State(app): State<Shared>,
who: Option<axum::Extension<crate::server::Who>>,
Json(q): Json<api::RemoteListRequest>,
) -> Reply<api::RemoteList> {
let server = app.server.clone();
let run = async {
let source = crate::hosting::parse(&q.url)?;
let given = q
.token
.filter(|t| !t.trim().is_empty())
.map(|t| worktree::Auth {
username: q
.username
.filter(|u| !u.trim().is_empty())
.unwrap_or_else(|| worktree::token_user(&source.host).into()),
token: t.trim().to_string(),
});
let host = source.host.clone();
let auth = match given.clone() {
Some(a) => Some(a),
None => {
let server = server.clone();
tokio::task::spawn_blocking(move || {
let local = server.is_none();
let token = crate::config::git_credentials(server).token(&host);
match token {
Some(token) => Some(worktree::Auth {
username: worktree::token_user(&host).into(),
token,
}),
None if local => worktree::helper_auth(&host),
None => None,
}
})
.await?
}
};
let repos = crate::hosting::list(&source, auth.as_ref()).await?;
let mut warning = None;
if let Some(a) = given {
let host = source.host.clone();
match (&server, who) {
(Some(server), Some(w)) => {
if let Err(e) = server.secrets.set_secret(
&w.email,
&crate::server::git_secret(&host),
&a.token,
) {
warning = Some(format!("the token worked but was not kept: {e:#}"));
}
}
_ => tokio::task::spawn_blocking(move || worktree::approve(&host, &a)).await?,
}
}
let here: Vec<String> = app
.repos()
.iter()
.filter_map(|r| {
r.remote
.clone()
.or_else(|| worktree::remote_url(Path::new(&r.path)))
})
.map(|u| same_repo(&u))
.collect();
let rows = repos
.into_iter()
.map(|repo| api::RemoteRow {
listed: here.contains(&same_repo(&repo.clone_url)),
repo,
})
.collect();
anyhow::Ok(api::RemoteList {
ok: true,
provider: source.provider,
host: source.host,
owner: source.owner,
repos: rows,
warning,
})
};
run.await
.map(Json)
.map_err(|e| match e.downcast_ref::<worktree::NeedsToken>() {
Some(n) => {
let error = api::ErrorDetail {
code: "needs_token".into(),
message: e.to_string(),
host: Some(n.host.clone()),
username: Some(n.username.clone()),
};
super::Failure(StatusCode::UNAUTHORIZED, error)
}
None => fail(e),
})
}
async fn label_added(
app: &Shared,
who: &super::Member,
names: &[String],
teams: &[String],
) -> anyhow::Result<()> {
if teams.is_empty() || names.is_empty() {
return Ok(());
}
super::team::change_teams(app, who, None, |t| {
for n in names {
t.label(n, teams);
}
Ok(())
})
.await
}
pub(super) async fn remote_add(
who: super::Member,
State(app): State<Shared>,
Cur(s): Cur,
Json(q): Json<api::RemoteAddRequest>,
) -> Reply<api::Done> {
let run = async {
let mut list = app.repos();
let mut added = 0;
let mut named = Vec::new();
for url in &q.urls {
let path = crate::hosting::managed_path(Path::new(&crate::config::data_dir()), url)?
.to_string_lossy()
.into_owned();
named.push(worktree::name(Path::new(&path)));
if !list.iter().any(|r| r.path == path) {
list.push(RepoEntry {
path,
remote: Some(url.clone()),
});
added += 1;
}
}
app.set_repos(who.as_ref().map(|w| w.email.as_str()), &list)?;
label_added(&app, &who, &named, &q.teams).await?;
let mut shared = 0;
if q.share {
let root = s.data.root.clone().context("no index to share with")?;
let stored = crate::team::load(&root).await?;
let mut settings = stored.settings;
for url in &q.urls {
if !settings
.repos
.iter()
.any(|u| same_repo(u) == same_repo(url))
{
settings.repos.push(url.clone());
shared += 1;
}
}
if shared > 0 {
crate::config::save_team(&root, &settings, stored.version).await?;
app.set_team(settings);
}
}
anyhow::Ok(match (added, q.share) {
(_, true) => format!(
"added {added}, {shared} new for the org; press Index now to clone and index them"
),
_ => format!("added {added}; press Index now to clone and index them"),
})
};
run.await.map(|m| Json(api::Done::new(m))).map_err(fail)
}