69f12aa
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::sync::queue::SyncQueue;
use config::git::SyncSettings;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PostReceiveRefUpdate {
pub name: String,
pub old_oid: String,
pub new_oid: String,
}
impl PostReceiveRefUpdate {
#[must_use]
pub fn new(
name: impl Into<String>,
old_oid: impl Into<String>,
new_oid: impl Into<String>,
) -> Self {
Self {
name: name.into(),
old_oid: old_oid.into(),
new_oid: new_oid.into(),
}
}
#[must_use]
pub fn is_delete(&self) -> bool {
self.is_zero(self.new_oid.as_str())
}
#[must_use]
pub fn is_create(&self) -> bool {
self.is_zero(self.old_oid.as_str()) && !self.is_zero(self.new_oid.as_str())
}
fn is_zero(&self, oid: &str) -> bool {
oid.is_empty() || oid.chars().all(|c| c == '0')
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PostReceiveTask {
pub id: Uuid,
pub repo_id: Uuid,
pub channel_id: Uuid,
pub actor_id: Option<i64>,
pub refs: Vec<PostReceiveRefUpdate>,
#[serde(default)]
pub trace: Option<String>,
#[serde(default)]
pub request_id: Option<String>,
}
impl PostReceiveTask {
#[must_use]
pub fn new(
repo_id: Uuid,
channel_id: Uuid,
actor_id: Option<i64>,
refs: Vec<PostReceiveRefUpdate>,
) -> Self {
Self {
id: Uuid::now_v7(),
repo_id,
channel_id,
actor_id,
refs,
trace: export::telemetry::traceparent(),
request_id: export::telemetry::request_id(),
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum EnqueueError {
#[error("serialize task: {0}")]
Serialize(#[from] serde_json::Error),
#[error("queue error: {0}")]
Queue(String),
}
pub async fn enqueue_post_receive(
queue: &SyncQueue,
settings: &SyncSettings,
task: &PostReceiveTask,
) -> Result<(), EnqueueError> {
let payload = serde_json::to_vec(task)?;
let queue_len = queue
.publish_raw(
&settings.queue_topic,
Some(task.repo_id.as_bytes()),
&payload,
)
.await
.map_err(|e| EnqueueError::Queue(e.to_string()))?;
tracing::info!(
task_id = %task.id,
repo_id = %task.repo_id,
refs = task.refs.len(),
topic = %settings.queue_topic,
queue_len,
"sync task enqueued"
);
Ok(())
}
pub async fn enqueue_post_receive_dlq(
queue: &SyncQueue,
settings: &SyncSettings,
task: &PostReceiveTask,
) -> Result<(), EnqueueError> {
let payload = serde_json::to_vec(task)?;
queue
.publish_raw(&settings.dlq_topic, Some(task.repo_id.as_bytes()), &payload)
.await
.map_err(|e| EnqueueError::Queue(e.to_string()))?;
tracing::debug!(
task_id = %task.id,
repo_id = %task.repo_id,
topic = %settings.dlq_topic,
"task enqueued to dlq"
);
Ok(())
}