d3d11f4
use std::error::Error;
use std::sync::Arc;
use export::AppDatabase;
use models::repo::repos;
use models::space::spaces;
use models::webhook::repo_webhook_delivery::{self, RepoWebhookDeliveryStatus};
use models::webhook::repo_webhooks;
use models::webhook::webhooks::WebhookStatus;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::sync::event::{PostReceiveRefUpdate, PostReceiveTask};
const PUSH_EVENT: &str = "push";
#[async_trait::async_trait]
pub trait WebhookTx: Send + Sync {
async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>>;
}
#[derive(Debug, Default, Clone)]
pub struct NoopWebhookTx;
#[async_trait::async_trait]
impl WebhookTx for NoopWebhookTx {
async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>> {
tracing::warn!(
task_id = %task.id,
repo_id = %task.repo_id,
refs = task.refs.len(),
"webhook dispatch not configured; message dropped (NoopWebhookTx)"
);
Ok(())
}
}
#[derive(Clone)]
pub struct RepoWebhookTx {
db: Arc<AppDatabase>,
}
impl RepoWebhookTx {
#[must_use]
pub fn new(db: Arc<AppDatabase>) -> Self {
Self { db }
}
}
#[async_trait::async_trait]
impl WebhookTx for RepoWebhookTx {
async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>> {
if task.refs.is_empty() {
return Ok(());
}
let hooks = repo_webhooks::Entity::find()
.filter(repo_webhooks::Column::RepoId.eq(task.repo_id))
.filter(repo_webhooks::Column::Status.eq(WebhookStatus::Active))
.all(self.db.read())
.await?;
if hooks.is_empty() {
return Ok(());
}
let Some(repo) = repos::Entity::find_by_id(task.repo_id)
.one(self.db.read())
.await?
else {
tracing::warn!(repo_id = %task.repo_id, "repo missing for webhook dispatch");
return Ok(());
};
let space = spaces::Entity::find_by_id(task.channel_id)
.one(self.db.read())
.await?;
let full_name = match &space {
Some(space) => format!("{}/{}", space.lower_name, repo.name),
None => repo.name.clone(),
};
let now = OffsetDateTime::now_utc();
let mut enqueued = 0u64;
for hook in hooks {
if !subscribes_push(&hook.events) {
continue;
}
for update in &task.refs {
let payload = push_payload(task, &repo, &full_name, update);
repo_webhook_delivery::ActiveModel {
id: Set(Uuid::now_v7()),
webhook_id: Set(hook.id),
repo_id: Set(task.repo_id),
event_type: Set(PUSH_EVENT.to_string()),
payload: Set(payload),
status: Set(RepoWebhookDeliveryStatus::Pending),
attempts: Set(0),
next_retry_at: Set(now),
response_status: Set(None),
response_body: Set(None),
last_error: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db.write())
.await?;
enqueued += 1;
}
}
if enqueued > 0 {
tracing::debug!(
task_id = %task.id,
repo_id = %task.repo_id,
deliveries = enqueued,
"push webhook deliveries enqueued"
);
}
Ok(())
}
}
fn subscribes_push(events: &serde_json::Value) -> bool {
let events: Vec<String> = serde_json::from_value(events.clone()).unwrap_or_default();
events
.iter()
.any(|event| event == PUSH_EVENT || event == "*")
}
fn push_payload(
task: &PostReceiveTask,
repo: &repos::Model,
full_name: &str,
update: &PostReceiveRefUpdate,
) -> serde_json::Value {
let ref_type = if update.name.starts_with("refs/tags/") {
"tag"
} else if update.name.starts_with("refs/heads/") {
"branch"
} else {
"other"
};
let actor = task.actor_id.map(|id| id.to_string());
serde_json::json!({
"ref": update.name,
"before": update.old_oid,
"after": update.new_oid,
"created": update.is_create(),
"deleted": update.is_delete(),
"ref_type": ref_type,
"pusher": { "id": actor },
"sender": { "id": actor },
"repository": {
"id": repo.id,
"name": repo.name,
"full_name": full_name,
"private": repo.is_private,
"default_branch": repo.default_branch,
"space_id": task.channel_id,
},
})
}