feat: Complete collaboration event delivery #3
+25
-3
@@ -1,190 +1,212 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use hmac::{Hmac, KeyInit, Mac};
|
||||
use sha2::Sha256;
|
||||
use syncode_collab_application::{
|
||||
EventPublisher, OutboxMessage, OutboxQueue, PublishError, dispatch,
|
||||
};
|
||||
use syncode_collab_model::envelope::PROTOCOL_VERSION;
|
||||
use syncode_collab_model::{Envelope, MessageType, SchemaVersion};
|
||||
use uuid::Uuid;
|
||||
|
||||
pub struct OutboxWorker<Q> {
|
||||
queue: Q,
|
||||
publisher: Publisher,
|
||||
interval: Duration,
|
||||
batch: i64,
|
||||
lease: i64,
|
||||
}
|
||||
|
||||
pub struct OutboxWorkerConfig {
|
||||
pub source_node_id: Uuid,
|
||||
pub control_url: String,
|
||||
pub secret: String,
|
||||
pub interval: Duration,
|
||||
pub batch: i64,
|
||||
pub lease: i64,
|
||||
}
|
||||
|
||||
impl<Q> OutboxWorker<Q>
|
||||
where
|
||||
Q: OutboxQueue,
|
||||
{
|
||||
pub fn new(queue: Q, config: OutboxWorkerConfig) -> Result<Self, String> {
|
||||
if config.interval.is_zero() || config.batch <= 0 || config.lease <= 0 {
|
||||
return Err("outbox interval, batch and lease must be positive".to_owned());
|
||||
}
|
||||
let control = ControlPublisher::new(config.control_url, config.secret)?;
|
||||
Ok(Self {
|
||||
queue,
|
||||
publisher: Publisher {
|
||||
source_node_id: config.source_node_id,
|
||||
control,
|
||||
},
|
||||
interval: config.interval,
|
||||
batch: config.batch,
|
||||
lease: config.lease,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn run(self) -> Result<(), String> {
|
||||
loop {
|
||||
dispatch(&self.queue, &self.publisher, self.batch, self.lease)
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
tokio::time::sleep(self.interval).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct Publisher {
|
||||
source_node_id: Uuid,
|
||||
control: ControlPublisher,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventPublisher for Publisher {
|
||||
async fn publish(&self, message: &OutboxMessage) -> Result<(), PublishError> {
|
||||
let envelope = envelope(self.source_node_id, message)?;
|
||||
if control_event(message) {
|
||||
self.control.publish(&envelope).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn envelope(source_node_id: Uuid, message: &OutboxMessage) -> Result<Envelope, PublishError> {
|
||||
Ok(Envelope {
|
||||
message_id: message.message_id,
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
schema_version: SchemaVersion(1),
|
||||
source_node_id: source_node_id.to_string(),
|
||||
repository_id: message.resource_id,
|
||||
message_type: MessageType::new(message.message_type.clone())
|
||||
.map_err(|error| PublishError::Rejected(error.to_string()))?,
|
||||
sequence: message.id,
|
||||
term: 1,
|
||||
payload: message.payload.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
fn control_event(message: &OutboxMessage) -> bool {
|
||||
matches!(
|
||||
message.message_type.as_str(),
|
||||
"collaboration.pull_request.created"
|
||||
| "collaboration.pull_request.synchronized"
|
||||
| "collaboration.pull_request.state_changed"
|
||||
) && message.payload.get("head_oid").is_some()
|
||||
}
|
||||
|
||||
struct ControlPublisher {
|
||||
client: reqwest::Client,
|
||||
url: reqwest::Url,
|
||||
secret: String,
|
||||
}
|
||||
|
||||
impl ControlPublisher {
|
||||
fn new(url: String, secret: String) -> Result<Self, String> {
|
||||
if secret.is_empty() {
|
||||
return Err("event secret must not be empty".to_owned());
|
||||
}
|
||||
let url = reqwest::Url::parse(&url).map_err(|error| error.to_string())?;
|
||||
let client = reqwest::Client::builder()
|
||||
.redirect(reqwest::redirect::Policy::none())
|
||||
.timeout(Duration::from_secs(10))
|
||||
.build()
|
||||
.map_err(|error| error.to_string())?;
|
||||
Ok(Self {
|
||||
client,
|
||||
url,
|
||||
secret,
|
||||
})
|
||||
}
|
||||
|
||||
async fn publish(&self, envelope: &Envelope) -> Result<(), PublishError> {
|
||||
let body = serde_json::to_vec(envelope)
|
||||
.map_err(|error| PublishError::Rejected(error.to_string()))?;
|
||||
let signature = signature(&self.secret, &body)?;
|
||||
let response = self
|
||||
.client
|
||||
.post(self.url.clone())
|
||||
.header("x-syncode-event", envelope.message_type.as_str())
|
||||
.header("x-syncode-delivery", envelope.message_id.to_string())
|
||||
.header("x-syncode-signature", signature)
|
||||
.body(body)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|error| PublishError::Unreachable(error.to_string()))?;
|
||||
if response.status().is_success() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(PublishError::Rejected(format!(
|
||||
"control returned HTTP {}",
|
||||
response.status()
|
||||
)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn signature(secret: &str, body: &[u8]) -> Result<String, PublishError> {
|
||||
let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes())
|
||||
.map_err(|error| PublishError::Rejected(error.to_string()))?;
|
||||
mac.update(body);
|
||||
Ok(hex::encode(mac.finalize().into_bytes()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use syncode_collab_application::OutboxMessage;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::control_event;
|
||||
|
||||
fn message(message_type: &str, payload: serde_json::Value) -> OutboxMessage {
|
||||
OutboxMessage {
|
||||
id: 1,
|
||||
message_id: Uuid::nil(),
|
||||
message_type: message_type.to_owned(),
|
||||
resource_id: Uuid::nil(),
|
||||
payload,
|
||||
attempts: 0,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sends_only_actionable_pull_request_facts_to_control() {
|
||||
assert!(control_event(&message(
|
||||
"collaboration.pull_request.created",
|
||||
serde_json::json!({"head_oid": "abc"}),
|
||||
)));
|
||||
assert!(!control_event(&message(
|
||||
"collaboration.pull_request.created",
|
||||
serde_json::json!({"number": 1}),
|
||||
)));
|
||||
assert!(!control_event(&message(
|
||||
"collaboration.pull_request.merged",
|
||||
serde_json::json!({"head_oid": "abc"}),
|
||||
)));
|
||||
}
|
||||
}
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use hmac::{Hmac, KeyInit, Mac};
|
||||
use sha2::Sha256;
|
||||
use syncode_collab_application::{
|
||||
EventPublisher, OutboxMessage, OutboxQueue, PublishError, dispatch,
|
||||
};
|
||||
use syncode_collab_model::envelope::PROTOCOL_VERSION;
|
||||
use syncode_collab_model::{Envelope, MessageType, SchemaVersion};
|
||||
use uuid::Uuid;
|
||||
|
||||
pub struct OutboxWorker<Q> {
|
||||
queue: Q,
|
||||
publisher: Publisher,
|
||||
interval: Duration,
|
||||
batch: i64,
|
||||
lease: i64,
|
||||
}
|
||||
|
||||
pub struct OutboxWorkerConfig {
|
||||
pub source_node_id: Uuid,
|
||||
pub control_url: String,
|
||||
pub secret: String,
|
||||
pub interval: Duration,
|
||||
pub batch: i64,
|
||||
pub lease: i64,
|
||||
}
|
||||
|
||||
impl<Q> OutboxWorker<Q>
|
||||
where
|
||||
Q: OutboxQueue,
|
||||
{
|
||||
pub fn new(queue: Q, config: OutboxWorkerConfig) -> Result<Self, String> {
|
||||
if config.interval.is_zero() || config.batch <= 0 || config.lease <= 0 {
|
||||
return Err("outbox interval, batch and lease must be positive".to_owned());
|
||||
}
|
||||
let control = ControlPublisher::new(config.control_url, config.secret)?;
|
||||
Ok(Self {
|
||||
queue,
|
||||
publisher: Publisher {
|
||||
source_node_id: config.source_node_id,
|
||||
control,
|
||||
},
|
||||
interval: config.interval,
|
||||
batch: config.batch,
|
||||
lease: config.lease,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn run(self) -> Result<(), String> {
|
||||
loop {
|
||||
dispatch(&self.queue, &self.publisher, self.batch, self.lease)
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
tokio::time::sleep(self.interval).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct Publisher {
|
||||
source_node_id: Uuid,
|
||||
control: ControlPublisher,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventPublisher for Publisher {
|
||||
async fn publish(&self, message: &OutboxMessage) -> Result<(), PublishError> {
|
||||
let envelope = envelope(self.source_node_id, message)?;
|
||||
if control_event(message) {
|
||||
self.control.publish(&envelope).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn envelope(source_node_id: Uuid, message: &OutboxMessage) -> Result<Envelope, PublishError> {
|
||||
Ok(Envelope {
|
||||
message_id: message.message_id,
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
schema_version: SchemaVersion(1),
|
||||
source_node_id: source_node_id.to_string(),
|
||||
repository_id: message.resource_id,
|
||||
message_type: MessageType::new(message.message_type.clone())
|
||||
.map_err(|error| PublishError::Rejected(error.to_string()))?,
|
||||
sequence: message.id,
|
||||
term: 1,
|
||||
payload: message.payload.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
fn control_event(message: &OutboxMessage) -> bool {
|
||||
matches!(
|
||||
message.message_type.as_str(),
|
||||
"collaboration.pull_request.created"
|
||||
| "collaboration.pull_request.synchronized"
|
||||
| "collaboration.pull_request.state_changed"
|
||||
) && [
|
||||
"repository_id",
|
||||
"head_repository_id",
|
||||
"number",
|
||||
"base_branch",
|
||||
"base_oid",
|
||||
"head_branch",
|
||||
"head_oid",
|
||||
"actor_id",
|
||||
"state",
|
||||
]
|
||||
.iter()
|
||||
.all(|field| message.payload.get(field).is_some())
|
||||
}
|
||||
|
||||
struct ControlPublisher {
|
||||
client: reqwest::Client,
|
||||
url: reqwest::Url,
|
||||
secret: String,
|
||||
}
|
||||
|
||||
impl ControlPublisher {
|
||||
fn new(url: String, secret: String) -> Result<Self, String> {
|
||||
if secret.is_empty() {
|
||||
return Err("event secret must not be empty".to_owned());
|
||||
}
|
||||
let url = reqwest::Url::parse(&url).map_err(|error| error.to_string())?;
|
||||
let client = reqwest::Client::builder()
|
||||
.redirect(reqwest::redirect::Policy::none())
|
||||
.timeout(Duration::from_secs(10))
|
||||
.build()
|
||||
.map_err(|error| error.to_string())?;
|
||||
Ok(Self {
|
||||
client,
|
||||
url,
|
||||
secret,
|
||||
})
|
||||
}
|
||||
|
||||
async fn publish(&self, envelope: &Envelope) -> Result<(), PublishError> {
|
||||
let body = serde_json::to_vec(envelope)
|
||||
.map_err(|error| PublishError::Rejected(error.to_string()))?;
|
||||
let signature = signature(&self.secret, &body)?;
|
||||
let response = self
|
||||
.client
|
||||
.post(self.url.clone())
|
||||
.header("x-syncode-event", envelope.message_type.as_str())
|
||||
.header("x-syncode-delivery", envelope.message_id.to_string())
|
||||
.header("x-syncode-signature", signature)
|
||||
.body(body)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|error| PublishError::Unreachable(error.to_string()))?;
|
||||
if response.status().is_success() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(PublishError::Rejected(format!(
|
||||
"control returned HTTP {}",
|
||||
response.status()
|
||||
)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn signature(secret: &str, body: &[u8]) -> Result<String, PublishError> {
|
||||
let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes())
|
||||
.map_err(|error| PublishError::Rejected(error.to_string()))?;
|
||||
mac.update(body);
|
||||
Ok(hex::encode(mac.finalize().into_bytes()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use syncode_collab_application::OutboxMessage;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::control_event;
|
||||
|
||||
fn message(message_type: &str, payload: serde_json::Value) -> OutboxMessage {
|
||||
OutboxMessage {
|
||||
id: 1,
|
||||
message_id: Uuid::nil(),
|
||||
message_type: message_type.to_owned(),
|
||||
resource_id: Uuid::nil(),
|
||||
payload,
|
||||
attempts: 0,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sends_only_actionable_pull_request_facts_to_control() {
|
||||
assert!(control_event(&message(
|
||||
"collaboration.pull_request.created",
|
||||
serde_json::json!({
|
||||
"repository_id": "repository",
|
||||
"head_repository_id": "head-repository",
|
||||
"number": 1,
|
||||
"base_branch": "main",
|
||||
"base_oid": "base",
|
||||
"head_branch": "feature",
|
||||
"head_oid": "head",
|
||||
"actor_id": "actor",
|
||||
"state": "open"
|
||||
}),
|
||||
)));
|
||||
assert!(!control_event(&message(
|
||||
"collaboration.pull_request.created",
|
||||
serde_json::json!({"head_oid": "head"}),
|
||||
)));
|
||||
assert!(!control_event(&message(
|
||||
"collaboration.pull_request.merged",
|
||||
serde_json::json!({"head_oid": "abc"}),
|
||||
)));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user