MaigoLabs/refray
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[O] Rework webhook

Browse Source
This commit is contained in:
Azalea Gui
2026-05-08 04:13:45 +00:00
parent 0566e97c6a
commit 67dd55a1cf
5 changed files with 431 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+18
View File
@@ -135,6 +135,12 @@ Expose that listener with your reverse proxy or tunnel, then install repository
git-sync webhook install
```
Manual `webhook install` always checks the selected repositories on the provider and repairs or records the hook state. To install or repair one repository exactly:
```sh
git-sync webhook install important-repo
```
You can also pass them explicitly:
```sh
@@ -163,6 +169,18 @@ To uninstall webhooks previously installed by `git-sync`:
git-sync webhook uninstall
```
Manual `webhook uninstall` checks repositories on the provider instead of trusting only local state. To uninstall one repository exactly:
```sh
git-sync webhook uninstall important-repo
```
To move installed hooks to a new public URL, use `webhook update`. It removes hooks matching the current configured `[webhook].url`, installs the new URL, updates `[webhook].url` in the config, and refreshes local webhook state:
```sh
git-sync webhook update --url https://new.example.com/webhook
```
Serve can also run periodic full syncs. The interval can be configured in `[webhook].full_sync_interval_minutes` or overridden at startup:
```sh
src/main.rs
+63 -2
View File
@@ -16,8 +16,8 @@ use clap::{Args, Parser, Subcommand};
use crate::config::{Config, default_config_path};
use crate::sync::{DEFAULT_JOBS, SyncOptions, sync_all};
use crate::webhook::{
ServeOptions, WebhookInstallOptions, WebhookUninstallOptions, install_webhooks, serve,
uninstall_webhooks,
ServeOptions, WebhookInstallOptions, WebhookUninstallOptions, WebhookUpdateOptions,
install_webhooks, serve, uninstall_webhooks, update_webhooks,
};
#[derive(Parser, Debug)]
@@ -84,10 +84,13 @@ struct ServeCommand {
enum WebhookCommand {
Install(WebhookInstallCommand),
Uninstall(WebhookUninstallCommand),
Update(WebhookUpdateCommand),
}
#[derive(Args, Debug)]
struct WebhookInstallCommand {
#[arg(value_name = "REPO", conflicts_with = "repo_pattern")]
repo: Option<String>,
#[arg(long, value_name = "URL")]
url: Option<String>,
#[arg(long, conflicts_with = "secret_env")]
@@ -108,6 +111,10 @@ struct WebhookInstallCommand {
#[derive(Args, Debug)]
struct WebhookUninstallCommand {
#[arg(value_name = "REPO")]
repo: Option<String>,
#[arg(long, value_name = "URL")]
url: Option<String>,
#[arg(long, value_name = "NAME")]
group: Option<String>,
#[arg(long)]
@@ -118,6 +125,22 @@ struct WebhookUninstallCommand {
jobs: usize,
}
#[derive(Args, Debug)]
struct WebhookUpdateCommand {
#[arg(long, value_name = "URL")]
url: String,
#[arg(long, conflicts_with = "secret_env")]
secret: Option<String>,
#[arg(long, value_name = "ENV", conflicts_with = "secret")]
secret_env: Option<String>,
#[arg(long)]
dry_run: bool,
#[arg(long, value_name = "PATH")]
work_dir: Option<PathBuf>,
#[arg(long, default_value_t = DEFAULT_JOBS, value_name = "N")]
jobs: usize,
}
fn main() -> Result<()> {
let cli = Cli::parse();
let config_path = cli.config.unwrap_or_else(default_config_path);
@@ -177,6 +200,7 @@ fn main() -> Result<()> {
url,
secret,
group: command.group,
repo: command.repo,
repo_pattern: command.repo_pattern,
dry_run: command.dry_run,
work_dir: command.work_dir,
@@ -186,16 +210,46 @@ fn main() -> Result<()> {
}
Command::Webhook(WebhookCommand::Uninstall(command)) => {
let config = load_config(&config_path)?;
let url = resolve_webhook_url(&config, command.url)?;
uninstall_webhooks(
&config,
WebhookUninstallOptions {
url,
group: command.group,
repo: command.repo,
dry_run: command.dry_run,
work_dir: command.work_dir,
jobs: command.jobs,
},
)
}
Command::Webhook(WebhookCommand::Update(command)) => {
let mut config = load_config(&config_path)?;
let old_url = config
.webhook
.as_ref()
.map(|webhook| webhook.url.clone())
.ok_or_else(|| {
anyhow::anyhow!("configure [webhook] before running webhook update")
})?;
let secret = resolve_webhook_secret(&config, command.secret, command.secret_env)?;
update_webhooks(
&config,
WebhookUpdateOptions {
old_url,
new_url: command.url.clone(),
secret,
dry_run: command.dry_run,
work_dir: command.work_dir,
jobs: command.jobs,
},
)?;
if !command.dry_run {
set_config_webhook_url(&mut config, command.url);
config.save(&config_path)?;
}
Ok(())
}
}
}
@@ -228,6 +282,13 @@ fn resolve_webhook_url(config: &Config, value: Option<String>) -> Result<String>
.ok_or_else(|| anyhow::anyhow!("pass --url or configure [webhook].url"))
}
fn set_config_webhook_url(config: &mut Config, url: String) {
let Some(webhook) = &mut config.webhook else {
unreachable!("caller verifies webhook config exists before saving update")
};
webhook.url = url;
}
#[cfg(test)]
#[path = "../tests/unit/cli.rs"]
mod tests;
src/webhook.rs
+142 -45
View File
@@ -40,6 +40,7 @@ pub struct WebhookInstallOptions {
pub url: String,
pub secret: String,
pub group: Option<String>,
pub repo: Option<String>,
pub repo_pattern: Option<String>,
pub dry_run: bool,
pub work_dir: Option<PathBuf>,
@@ -48,7 +49,19 @@ pub struct WebhookInstallOptions {
#[derive(Clone, Debug)]
pub struct WebhookUninstallOptions {
pub url: String,
pub group: Option<String>,
pub repo: Option<String>,
pub dry_run: bool,
pub work_dir: Option<PathBuf>,
pub jobs: usize,
}
#[derive(Clone, Debug)]
pub struct WebhookUpdateOptions {
pub old_url: String,
pub new_url: String,
pub secret: String,
pub dry_run: bool,
pub work_dir: Option<PathBuf>,
pub jobs: usize,
@@ -204,6 +217,9 @@ pub fn install_webhooks(config: &Config, options: WebhookInstallOptions) -> Resu
.list_repos(endpoint)
.with_context(|| format!("failed to list repos for {}", endpoint.label()))?;
for repo in repos {
if options.repo.as_ref().is_some_and(|name| name != &repo.name) {
continue;
}
if repo_pattern
.as_ref()
.is_some_and(|pattern| !pattern.is_match(&repo.name))
@@ -221,7 +237,7 @@ pub fn install_webhooks(config: &Config, options: WebhookInstallOptions) -> Resu
});
}
}
run_install_tasks(tasks, options.jobs, Arc::clone(&state))?;
run_install_tasks(tasks, options.jobs, Arc::clone(&state), false)?;
}
if !options.dry_run {
let state = state
@@ -239,41 +255,99 @@ pub fn uninstall_webhooks(config: &Config, options: WebhookUninstallOptions) ->
}
let work_dir = options.work_dir.clone().unwrap_or_else(default_work_dir);
let mut state = load_webhook_state(&work_dir)?;
if state.installations.is_empty() {
crate::logln!(
"{} no webhook installations recorded",
style("skip").yellow().bold()
);
return Ok(());
}
let mut tasks = Vec::new();
for (key, installation) in &state.installations {
for mirror in &config.mirrors {
if options
.group
.as_ref()
.is_some_and(|group| group != &installation.group)
.is_some_and(|group| group != &mirror.name)
{
continue;
}
tasks.push(WebhookUninstallTask {
key: key.clone(),
site: config.site(&installation.endpoint.site).cloned(),
installation: installation.clone(),
dry_run: options.dry_run,
});
crate::logln!();
crate::logln!(
"{} {}",
style("Webhook group").cyan().bold(),
style(&mirror.name).bold()
);
for endpoint in &mirror.endpoints {
let site = config.site(&endpoint.site).unwrap();
let client = ProviderClient::new(site)?;
crate::logln!(
" {} {}",
style("list").cyan().bold(),
style(endpoint.label()).dim()
);
let repos = client
.list_repos(endpoint)
.with_context(|| format!("failed to list repos for {}", endpoint.label()))?;
for repo in repos {
if options.repo.as_ref().is_some_and(|name| name != &repo.name) {
continue;
}
tasks.push(WebhookUninstallTask {
group: mirror.name.clone(),
site: site.clone(),
endpoint: endpoint.clone(),
repo,
url: options.url.clone(),
dry_run: options.dry_run,
});
}
}
}
let removed_keys = run_uninstall_tasks(tasks, options.jobs)?;
if !options.dry_run {
for key in removed_keys {
state.installations.remove(&key);
state.skipped.remove(&key);
}
save_webhook_state(&work_dir, &state)?;
}
Ok(())
}
pub fn update_webhooks(config: &Config, options: WebhookUpdateOptions) -> Result<()> {
validate_config(config)?;
if options.jobs == 0 {
bail!("--jobs must be at least 1");
}
if options.old_url != options.new_url {
crate::logln!(
"{} {} -> {}",
style("Webhook URL").cyan().bold(),
style(&options.old_url).dim(),
style(&options.new_url).cyan()
);
uninstall_webhooks(
config,
WebhookUninstallOptions {
url: options.old_url.clone(),
group: None,
repo: None,
dry_run: options.dry_run,
work_dir: options.work_dir.clone(),
jobs: options.jobs,
},
)?;
}
install_webhooks(
config,
WebhookInstallOptions {
url: options.new_url,
secret: options.secret,
group: None,
repo: None,
repo_pattern: None,
dry_run: options.dry_run,
work_dir: options.work_dir,
jobs: options.jobs,
},
)
}
pub fn ensure_configured_webhooks(
config: &Config,
mirror: &MirrorConfig,
@@ -307,7 +381,7 @@ pub fn ensure_configured_webhooks(
dry_run: false,
});
}
run_install_tasks(tasks, jobs, Arc::clone(&state))?;
run_install_tasks(tasks, jobs, Arc::clone(&state), true)?;
let state = state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
@@ -422,9 +496,11 @@ struct WebhookInstallTask {
#[derive(Clone, Debug)]
struct WebhookUninstallTask {
key: String,
site: Option<crate::config::SiteConfig>,
installation: WebhookInstallation,
group: String,
site: crate::config::SiteConfig,
endpoint: EndpointConfig,
repo: RemoteRepo,
url: String,
dry_run: bool,
}
@@ -432,6 +508,7 @@ fn run_install_tasks(
tasks: Vec<WebhookInstallTask>,
jobs: usize,
state: Arc<Mutex<WebhookState>>,
use_state_cache: bool,
) -> Result<()> {
if tasks.is_empty() {
return Ok(());
@@ -461,7 +538,7 @@ fn run_install_tasks(
break;
};
if result_sender
.send(install_webhook_task(task, &state))
.send(install_webhook_task(task, &state, use_state_cache))
.is_err()
{
break;
@@ -549,9 +626,13 @@ fn run_uninstall_tasks(tasks: Vec<WebhookUninstallTask>, jobs: usize) -> Result<
Ok(removed_keys)
}
fn install_webhook_task(task: WebhookInstallTask, state: &Arc<Mutex<WebhookState>>) -> Result<()> {
fn install_webhook_task(
task: WebhookInstallTask,
state: &Arc<Mutex<WebhookState>>,
use_state_cache: bool,
) -> Result<()> {
let key = webhook_installation_key(&task.group, &task.endpoint, &task.repo.name);
{
if use_state_cache {
let state = state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
@@ -588,6 +669,16 @@ fn install_webhook_task(task: WebhookInstallTask, state: &Arc<Mutex<WebhookState
let client = ProviderClient::new(&task.site)?;
if let Err(error) = client.install_webhook(&task.endpoint, &task.repo, &task.url, &task.secret)
{
if is_duplicate_webhook_error(&error) {
crate::logln!(
" {} {} {}",
style("exists").green().bold(),
style(&task.repo.name).cyan(),
style(format!("webhook on {}", task.endpoint.label())).dim()
);
record_webhook_installation(state, key, task);
return Ok(());
}
if let Some(reason) = non_actionable_webhook_failure_reason(&error) {
crate::logln!(
" {} {} {}",
@@ -618,6 +709,15 @@ fn install_webhook_task(task: WebhookInstallTask, state: &Arc<Mutex<WebhookState
)
});
}
record_webhook_installation(state, key, task);
Ok(())
}
fn record_webhook_installation(
state: &Arc<Mutex<WebhookState>>,
key: String,
task: WebhookInstallTask,
) {
let mut state = state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
@@ -631,10 +731,10 @@ fn install_webhook_task(task: WebhookInstallTask, state: &Arc<Mutex<WebhookState
url: task.url,
},
);
Ok(())
}
fn uninstall_webhook_task(task: WebhookUninstallTask) -> Result<Option<String>> {
let key = webhook_installation_key(&task.group, &task.endpoint, &task.repo.name);
crate::logln!(
" {} {} {}",
style(if task.dry_run {
@@ -644,36 +744,23 @@ fn uninstall_webhook_task(task: WebhookUninstallTask) -> Result<Option<String>>
})
.red()
.bold(),
style(&task.installation.repo).cyan(),
style(format!("from {}", task.installation.endpoint.label())).dim()
style(&task.repo.name).cyan(),
style(format!("from {}", task.endpoint.label())).dim()
);
if task.dry_run {
return Ok(None);
}
let Some(site) = task.site else {
crate::logln!(
" {} {} {}",
style("skip").yellow().bold(),
style(&task.installation.repo).cyan(),
style(format!("unknown site {}", task.installation.endpoint.site)).dim()
);
return Ok(None);
};
let client = ProviderClient::new(&site)?;
let client = ProviderClient::new(&task.site)?;
client
.uninstall_webhook(
&task.installation.endpoint,
&task.installation.repo,
&task.installation.url,
)
.uninstall_webhook(&task.endpoint, &task.repo.name, &task.url)
.with_context(|| {
format!(
"failed to uninstall webhook for {} from {}",
task.installation.repo,
task.installation.endpoint.label()
task.repo.name,
task.endpoint.label()
)
})?;
Ok(Some(task.key))
Ok(Some(key))
}
fn non_actionable_webhook_failure_reason(error: &anyhow::Error) -> Option<String> {
@@ -700,6 +787,16 @@ fn non_actionable_webhook_failure_reason(error: &anyhow::Error) -> Option<String
None
}
fn is_duplicate_webhook_error(error: &anyhow::Error) -> bool {
let text = error
.chain()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join("\n")
.to_ascii_lowercase();
text.contains("422 unprocessable entity") && text.contains("hook already exists")
}
fn webhook_installation_key(group: &str, endpoint: &EndpointConfig, repo: &str) -> String {
format!(
"{}\t{}\t{:?}\t{}\t{}",
tests/unit/cli.rs
+70 -3
View File
@@ -89,14 +89,13 @@ fn cli_accepts_webhook_install() {
"git-sync",
"webhook",
"install",
"repo-one",
"--url",
"https://mirror.example.test/webhook",
"--secret",
"secret",
"--group",
"sync-1",
"--repo-pattern",
"^repo$",
"--jobs",
"6",
])
@@ -111,16 +110,42 @@ fn cli_accepts_webhook_install() {
);
assert_eq!(args.secret, Some("secret".to_string()));
assert_eq!(args.group, Some("sync-1".to_string()));
assert_eq!(args.repo_pattern, Some("^repo$".to_string()));
assert_eq!(args.repo, Some("repo-one".to_string()));
assert_eq!(args.repo_pattern, None);
assert_eq!(args.jobs, 6);
}
#[test]
fn cli_accepts_webhook_install_repo_pattern() {
let cli = Cli::try_parse_from([
"git-sync",
"webhook",
"install",
"--url",
"https://mirror.example.test/webhook",
"--secret",
"secret",
"--repo-pattern",
"^repo$",
])
.unwrap();
let Command::Webhook(WebhookCommand::Install(args)) = cli.command else {
panic!("parsed unexpected command");
};
assert_eq!(args.repo, None);
assert_eq!(args.repo_pattern, Some("^repo$".to_string()));
}
#[test]
fn cli_accepts_webhook_uninstall() {
let cli = Cli::try_parse_from([
"git-sync",
"webhook",
"uninstall",
"repo-one",
"--url",
"https://mirror.example.test/webhook",
"--group",
"sync-1",
"--dry-run",
@@ -132,7 +157,49 @@ fn cli_accepts_webhook_uninstall() {
let Command::Webhook(WebhookCommand::Uninstall(args)) = cli.command else {
panic!("parsed unexpected command");
};
assert_eq!(
args.url,
Some("https://mirror.example.test/webhook".to_string())
);
assert_eq!(args.group, Some("sync-1".to_string()));
assert_eq!(args.repo, Some("repo-one".to_string()));
assert!(args.dry_run);
assert_eq!(args.jobs, 3);
}
#[test]
fn cli_accepts_webhook_update() {
let cli = Cli::try_parse_from([
"git-sync",
"webhook",
"update",
"--url",
"https://new.example.test/webhook",
"--secret-env",
"WEBHOOK_SECRET",
"--jobs",
"5",
])
.unwrap();
let Command::Webhook(WebhookCommand::Update(args)) = cli.command else {
panic!("parsed unexpected command");
};
assert_eq!(args.url, "https://new.example.test/webhook");
assert_eq!(args.secret_env, Some("WEBHOOK_SECRET".to_string()));
assert_eq!(args.jobs, 5);
}
#[test]
fn cli_rejects_scoped_webhook_update() {
let result = Cli::try_parse_from([
"git-sync",
"webhook",
"update",
"repo-one",
"--url",
"https://new.example.test/webhook",
]);
assert!(result.is_err());
}
tests/unit/webhook.rs
+138
View File
@@ -182,6 +182,7 @@ fn blocked_webhook_install_is_skipped_and_recorded() {
dry_run: false,
},
&state,
false,
)
.unwrap();
@@ -207,6 +208,115 @@ fn archived_webhook_failure_is_non_actionable() {
);
}
#[test]
fn duplicate_webhook_error_records_existing_installation() {
let error = anyhow::anyhow!(
"{}",
r#"POST https://api.github.com/repos/alice/repo/hooks returned 422 Unprocessable Entity: {"message":"Validation Failed","errors":[{"resource":"Hook","code":"custom","message":"Hook already exists on this repository"}]}"#
);
assert!(is_duplicate_webhook_error(&error));
let state = Arc::new(Mutex::new(WebhookState::default()));
let task = WebhookInstallTask {
site: site("github", ProviderKind::Github),
group: "sync-1".to_string(),
endpoint: endpoint("github", NamespaceKind::User, "alice"),
repo: RemoteRepo {
name: "repo".to_string(),
clone_url: "https://github.com/alice/repo.git".to_string(),
private: true,
description: None,
},
url: "https://mirror.example.test/webhook".to_string(),
secret: "secret".to_string(),
dry_run: false,
};
let key = webhook_installation_key(&task.group, &task.endpoint, &task.repo.name);
record_webhook_installation(&state, key.clone(), task);
let state = state.lock().unwrap();
assert!(state.skipped.is_empty());
assert_eq!(state.installations[&key].repo, "repo");
}
#[test]
fn install_task_state_cache_is_only_used_for_sync() {
let (api_url, handle) = request_server(
vec![("200 OK", "[]"), ("201 Created", r#"{"id":1}"#)],
|index, request| match index {
0 => assert!(request.starts_with("GET /repos/alice/repo/hooks ")),
1 => assert!(request.starts_with("POST /repos/alice/repo/hooks ")),
_ => unreachable!(),
},
);
let endpoint = endpoint("github", NamespaceKind::User, "alice");
let key = webhook_installation_key("sync-1", &endpoint, "repo");
let state = Arc::new(Mutex::new(WebhookState::default()));
state.lock().unwrap().installations.insert(
key,
WebhookInstallation {
group: "sync-1".to_string(),
endpoint: endpoint.clone(),
repo: "repo".to_string(),
url: "https://mirror.example.test/webhook".to_string(),
},
);
install_webhook_task(
WebhookInstallTask {
site: SiteConfig {
api_url: Some(api_url),
..site("github", ProviderKind::Github)
},
group: "sync-1".to_string(),
endpoint,
repo: RemoteRepo {
name: "repo".to_string(),
clone_url: "https://github.com/alice/repo.git".to_string(),
private: true,
description: None,
},
url: "https://mirror.example.test/webhook".to_string(),
secret: "secret".to_string(),
dry_run: false,
},
&state,
false,
)
.unwrap();
handle.join().unwrap();
}
#[test]
fn uninstall_task_removes_state_even_when_hook_is_missing() {
let (api_url, handle) = one_request_server("200 OK", "[]", |request| {
assert!(request.starts_with("GET /repos/alice/repo/hooks "))
});
let key = uninstall_webhook_task(WebhookUninstallTask {
group: "sync-1".to_string(),
site: SiteConfig {
api_url: Some(api_url),
..site("github", ProviderKind::Github)
},
endpoint: endpoint("github", NamespaceKind::User, "alice"),
repo: RemoteRepo {
name: "repo".to_string(),
clone_url: "https://github.com/alice/repo.git".to_string(),
private: true,
description: None,
},
url: "https://mirror.example.test/webhook".to_string(),
dry_run: false,
})
.unwrap();
assert_eq!(key.as_deref(), Some("sync-1\tgithub\tUser\talice\trepo"));
handle.join().unwrap();
}
fn site(name: &str, provider: ProviderKind) -> SiteConfig {
SiteConfig {
name: name.to_string(),
@@ -252,3 +362,31 @@ where
});
(format!("http://{address}"), handle)
}
fn request_server<F>(
responses: Vec<(&'static str, &'static str)>,
mut assert_request: F,
) -> (String, thread::JoinHandle<()>)
where
F: FnMut(usize, &str) + Send + 'static,
{
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let handle = thread::spawn(move || {
for (index, (status, body)) in responses.into_iter().enumerate() {
let (mut stream, _) = listener.accept().unwrap();
let mut buffer = [0_u8; 4096];
let bytes = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..bytes]).to_string();
assert_request(index, &request);
write!(
stream,
"HTTP/1.1 {status}\r\ncontent-type: application/json\r\nconnection: close\r\ncontent-length: {}\r\n\r\n{body}",
body.len()
)
.unwrap();
}
});
(format!("http://{address}"), handle)
}