feat: Complete collaboration event delivery #3
File diff suppressed because it is too large
Load Diff
@@ -1,66 +1,67 @@
|
||||
[package]
|
||||
name = "syncode-collab"
|
||||
description = "SynCode collaboration plane"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish = false
|
||||
|
||||
[workspace]
|
||||
members = [
|
||||
".",
|
||||
"crates/collab-api-graphql",
|
||||
"crates/collab-api-grpc",
|
||||
"crates/collab-application",
|
||||
"crates/collab-model",
|
||||
"crates/collab-store",
|
||||
]
|
||||
resolver = "3"
|
||||
|
||||
[workspace.package]
|
||||
version = "0.6.0"
|
||||
edition = "2024"
|
||||
rust-version = "1.95"
|
||||
license = "MIT"
|
||||
repository = "https://syncode.sh/syncode/collab"
|
||||
|
||||
[workspace.lints.rust]
|
||||
unsafe_code = "forbid"
|
||||
|
||||
[workspace.lints.clippy]
|
||||
expect_used = "deny"
|
||||
panic = "deny"
|
||||
unwrap_used = "deny"
|
||||
|
||||
[dependencies]
|
||||
async-graphql = "7.2.1"
|
||||
async-graphql-axum = "7.2.1"
|
||||
axum = "0.8.9"
|
||||
clap = { version = "4.6.4", features = ["derive", "env"] }
|
||||
hex = "0.4.3"
|
||||
hmac = "0.13.0"
|
||||
reqwest = { version = "0.13.2", default-features = false, features = ["json", "rustls"] }
|
||||
serde_json = "1.0.149"
|
||||
sha1 = "0.11.0"
|
||||
sha2 = "0.11.0"
|
||||
syncode-collab-api-graphql = { path = "crates/collab-api-graphql" }
|
||||
syncode-collab-api-grpc = { path = "crates/collab-api-grpc" }
|
||||
syncode-collab-application = { path = "crates/collab-application" }
|
||||
syncode-collab-model = { path = "crates/collab-model" }
|
||||
syncode-collab-store = { path = "crates/collab-store" }
|
||||
thiserror = "2.0.19"
|
||||
tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread", "signal", "time"] }
|
||||
tonic = "0.14.6"
|
||||
tower-http = { version = "0.6.8", features = ["cors"] }
|
||||
uuid = { version = "1.24.0", features = ["v4"] }
|
||||
|
||||
[dev-dependencies]
|
||||
syncode-collab-api-grpc = { path = "crates/collab-api-grpc" }
|
||||
tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread"] }
|
||||
tokio-stream = { version = "0.1.18", features = ["net"] }
|
||||
tonic = "0.14.6"
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
[package]
|
||||
name = "syncode-collab"
|
||||
description = "SynCode collaboration plane"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish = false
|
||||
|
||||
[workspace]
|
||||
members = [
|
||||
".",
|
||||
"crates/collab-api-graphql",
|
||||
"crates/collab-api-grpc",
|
||||
"crates/collab-application",
|
||||
"crates/collab-model",
|
||||
"crates/collab-store",
|
||||
]
|
||||
resolver = "3"
|
||||
|
||||
[workspace.package]
|
||||
version = "0.6.0"
|
||||
edition = "2024"
|
||||
rust-version = "1.95"
|
||||
license = "MIT"
|
||||
repository = "https://syncode.sh/syncode/collab"
|
||||
|
||||
[workspace.lints.rust]
|
||||
unsafe_code = "forbid"
|
||||
|
||||
[workspace.lints.clippy]
|
||||
expect_used = "deny"
|
||||
panic = "deny"
|
||||
unwrap_used = "deny"
|
||||
|
||||
[dependencies]
|
||||
async-trait = "0.1.89"
|
||||
async-graphql = "7.2.1"
|
||||
async-graphql-axum = "7.2.1"
|
||||
axum = "0.8.9"
|
||||
clap = { version = "4.6.4", features = ["derive", "env"] }
|
||||
hex = "0.4.3"
|
||||
hmac = "0.13.0"
|
||||
reqwest = { version = "0.13.2", default-features = false, features = ["json", "rustls"] }
|
||||
serde_json = "1.0.149"
|
||||
sha1 = "0.11.0"
|
||||
sha2 = "0.11.0"
|
||||
syncode-collab-api-graphql = { path = "crates/collab-api-graphql" }
|
||||
syncode-collab-api-grpc = { path = "crates/collab-api-grpc" }
|
||||
syncode-collab-application = { path = "crates/collab-application" }
|
||||
syncode-collab-model = { path = "crates/collab-model" }
|
||||
syncode-collab-store = { path = "crates/collab-store" }
|
||||
thiserror = "2.0.19"
|
||||
tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread", "signal", "time"] }
|
||||
tonic = "0.14.6"
|
||||
tower-http = { version = "0.6.8", features = ["cors"] }
|
||||
uuid = { version = "1.24.0", features = ["v4"] }
|
||||
|
||||
[dev-dependencies]
|
||||
syncode-collab-api-grpc = { path = "crates/collab-api-grpc" }
|
||||
tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread"] }
|
||||
tokio-stream = { version = "0.1.18", features = ["net"] }
|
||||
tonic = "0.14.6"
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
@@ -1,67 +1,91 @@
|
||||
use clap::Parser;
|
||||
|
||||
#[derive(Clone, Debug, Parser)]
|
||||
#[command(name = "syncode-collab", about = "SynCode collaboration plane")]
|
||||
pub struct Config {
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_LISTEN_GRPC",
|
||||
default_value = "0.0.0.0:8300"
|
||||
)]
|
||||
pub listen_grpc: String,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_LISTEN_HTTP",
|
||||
default_value = "0.0.0.0:8301"
|
||||
)]
|
||||
pub listen_http: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_DATABASE_URL")]
|
||||
pub database_url: String,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_DATABASE_MAX_CONNECTIONS",
|
||||
default_value_t = 16
|
||||
)]
|
||||
pub database_max_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
pub event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_WEBHOOK_KEY", hide_env_values = true)]
|
||||
pub webhook_key: String,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_WEBHOOK_INTERVAL_SECONDS",
|
||||
default_value_t = 2
|
||||
)]
|
||||
pub webhook_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_WEBHOOK_BATCH", default_value_t = 32)]
|
||||
pub webhook_batch: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
pub identity_grpc_url: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
pub identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
pub repository_grpc_url: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_GRPC_URL")]
|
||||
pub control_grpc_url: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_CORS_ORIGINS", value_delimiter = ',')]
|
||||
pub cors_origins: Vec<String>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_SESSION_COOKIE_NAME",
|
||||
default_value = "syncode_identity_session"
|
||||
)]
|
||||
pub session_cookie_name: String,
|
||||
}
|
||||
use clap::Parser;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Clone, Debug, Parser)]
|
||||
#[command(name = "syncode-collab", about = "SynCode collaboration plane")]
|
||||
pub struct Config {
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_LISTEN_GRPC",
|
||||
default_value = "0.0.0.0:8300"
|
||||
)]
|
||||
pub listen_grpc: String,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_LISTEN_HTTP",
|
||||
default_value = "0.0.0.0:8301"
|
||||
)]
|
||||
pub listen_http: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_DATABASE_URL")]
|
||||
pub database_url: String,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_DATABASE_MAX_CONNECTIONS",
|
||||
default_value_t = 16
|
||||
)]
|
||||
pub database_max_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
pub event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_SOURCE_NODE_ID")]
|
||||
pub source_node_id: Uuid,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_EVENT_URL")]
|
||||
pub control_event_url: String,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_OUTBOX_INTERVAL_SECONDS",
|
||||
default_value_t = 1
|
||||
)]
|
||||
pub outbox_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_OUTBOX_BATCH", default_value_t = 64)]
|
||||
pub outbox_batch: i64,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_OUTBOX_LEASE_SECONDS",
|
||||
default_value_t = 30
|
||||
)]
|
||||
pub outbox_lease_seconds: i64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_WEBHOOK_KEY", hide_env_values = true)]
|
||||
pub webhook_key: String,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_WEBHOOK_INTERVAL_SECONDS",
|
||||
default_value_t = 2
|
||||
)]
|
||||
pub webhook_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_WEBHOOK_BATCH", default_value_t = 32)]
|
||||
pub webhook_batch: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
pub identity_grpc_url: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
pub identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
pub repository_grpc_url: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_GRPC_URL")]
|
||||
pub control_grpc_url: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_CORS_ORIGINS", value_delimiter = ',')]
|
||||
pub cors_origins: Vec<String>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_COLLAB_SESSION_COOKIE_NAME",
|
||||
default_value = "syncode_identity_session"
|
||||
)]
|
||||
pub session_cookie_name: String,
|
||||
}
|
||||
@@ -1,12 +1,13 @@
|
||||
pub mod checks;
|
||||
pub mod config;
|
||||
pub mod http;
|
||||
pub mod identity;
|
||||
pub mod repository;
|
||||
pub mod runtime;
|
||||
mod webhook_format;
|
||||
mod webhook_request;
|
||||
mod webhook_worker;
|
||||
|
||||
pub use config::Config;
|
||||
pub use runtime::{RuntimeError, run};
|
||||
pub mod checks;
|
||||
pub mod config;
|
||||
pub mod http;
|
||||
pub mod identity;
|
||||
pub mod outbox_worker;
|
||||
pub mod repository;
|
||||
pub mod runtime;
|
||||
mod webhook_format;
|
||||
mod webhook_request;
|
||||
mod webhook_worker;
|
||||
|
||||
pub use config::Config;
|
||||
pub use runtime::{RuntimeError, run};
|
||||
+22
-2
@@ -1,142 +1,162 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use syncode_collab_api_grpc::CollabService;
|
||||
use syncode_collab_application::Collaboration;
|
||||
use syncode_collab_store::{PostgresCollaboration, PostgresInbox, connect, migrate};
|
||||
use tower_http::cors::{AllowOrigin, CorsLayer};
|
||||
|
||||
use crate::checks::ControlChecks;
|
||||
use crate::config::Config;
|
||||
use crate::http::router;
|
||||
use crate::identity::IdentityClientAccess;
|
||||
use crate::repository::RepositoryClientAccess;
|
||||
use crate::webhook_worker::WebhookWorker;
|
||||
|
||||
pub async fn run(config: Config) -> Result<(), RuntimeError> {
|
||||
if config.event_secret.is_empty() {
|
||||
return Err(RuntimeError::Configuration(
|
||||
"SYNCODE_COLLAB_EVENT_SECRET must not be empty".to_owned(),
|
||||
));
|
||||
}
|
||||
if config.identity_shared_secret.is_empty() {
|
||||
return Err(RuntimeError::Configuration(
|
||||
"SYNCODE_IDENTITY_SHARED_SECRET must not be empty".to_owned(),
|
||||
));
|
||||
}
|
||||
let webhook_key = parse_webhook_key(&config.webhook_key)?;
|
||||
let pool = connect(&config.database_url, config.database_max_connections)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Store(error.to_string()))?;
|
||||
migrate(&pool)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Store(error.to_string()))?;
|
||||
|
||||
let http = tokio::net::TcpListener::bind(&config.listen_http)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Listen(config.listen_http.clone(), error.to_string()))?;
|
||||
let grpc = config
|
||||
.listen_grpc
|
||||
.parse()
|
||||
.map_err(|_| RuntimeError::Address(config.listen_grpc.clone()))?;
|
||||
|
||||
let identity = Arc::new(
|
||||
IdentityClientAccess::connect(config.identity_grpc_url, config.identity_shared_secret)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Identity(error.to_string()))?,
|
||||
);
|
||||
let repository = Arc::new(
|
||||
RepositoryClientAccess::connect(config.repository_grpc_url)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Repository(error.to_string()))?,
|
||||
);
|
||||
let checks = Arc::new(
|
||||
ControlChecks::connect(config.control_grpc_url)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Control(error.to_string()))?,
|
||||
);
|
||||
let store = PostgresCollaboration::with_webhook_key(pool.clone(), webhook_key);
|
||||
let collaboration = Arc::new(Collaboration::with_delivery(
|
||||
Arc::new(store.clone()),
|
||||
identity,
|
||||
repository,
|
||||
checks,
|
||||
));
|
||||
let inbox = Arc::new(PostgresInbox::new(pool));
|
||||
let mut http_router = router(
|
||||
inbox,
|
||||
Arc::clone(&collaboration),
|
||||
config.event_secret,
|
||||
config.session_cookie_name,
|
||||
);
|
||||
if !config.cors_origins.is_empty() {
|
||||
let allowed = config.cors_origins;
|
||||
http_router = http_router.layer(
|
||||
CorsLayer::new()
|
||||
.allow_credentials(true)
|
||||
.allow_methods([
|
||||
axum::http::Method::GET,
|
||||
axum::http::Method::POST,
|
||||
axum::http::Method::OPTIONS,
|
||||
])
|
||||
.allow_headers([axum::http::header::CONTENT_TYPE])
|
||||
.allow_origin(AllowOrigin::predicate(move |origin, _| {
|
||||
origin
|
||||
.to_str()
|
||||
.is_ok_and(|origin| allowed.iter().any(|value| value == origin))
|
||||
})),
|
||||
);
|
||||
}
|
||||
let serve_http = axum::serve(http, http_router);
|
||||
let serve_grpc = tonic::transport::Server::builder()
|
||||
.add_service(CollabService::new(collaboration).into_server())
|
||||
.serve(grpc);
|
||||
let webhook_worker = WebhookWorker::new(
|
||||
store,
|
||||
Duration::from_secs(config.webhook_interval_seconds),
|
||||
config.webhook_batch,
|
||||
)
|
||||
.map_err(RuntimeError::Configuration)?
|
||||
.run();
|
||||
|
||||
tokio::select! {
|
||||
result = serve_http => result.map_err(|error| RuntimeError::Serve(error.to_string())),
|
||||
result = serve_grpc => result.map_err(|error| RuntimeError::Serve(error.to_string())),
|
||||
result = webhook_worker => result.map_err(RuntimeError::Webhook),
|
||||
_ = tokio::signal::ctrl_c() => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_webhook_key(value: &str) -> Result<[u8; 32], RuntimeError> {
|
||||
let decoded = hex::decode(value).map_err(|_| {
|
||||
RuntimeError::Configuration(
|
||||
"SYNCODE_COLLAB_WEBHOOK_KEY must be 64 hexadecimal characters".to_owned(),
|
||||
)
|
||||
})?;
|
||||
decoded.try_into().map_err(|_| {
|
||||
RuntimeError::Configuration(
|
||||
"SYNCODE_COLLAB_WEBHOOK_KEY must be 64 hexadecimal characters".to_owned(),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum RuntimeError {
|
||||
#[error("invalid collaboration configuration: {0}")]
|
||||
Configuration(String),
|
||||
#[error("{0} is not a valid socket address")]
|
||||
Address(String),
|
||||
#[error("cannot listen on {0}: {1}")]
|
||||
Listen(String, String),
|
||||
#[error("collaboration store failed: {0}")]
|
||||
Store(String),
|
||||
#[error("identity integration failed: {0}")]
|
||||
Identity(String),
|
||||
#[error("repository integration failed: {0}")]
|
||||
Repository(String),
|
||||
#[error("control integration failed: {0}")]
|
||||
Control(String),
|
||||
#[error("webhook delivery failed: {0}")]
|
||||
Webhook(String),
|
||||
#[error("serving failed: {0}")]
|
||||
Serve(String),
|
||||
}
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use syncode_collab_api_grpc::CollabService;
|
||||
use syncode_collab_application::Collaboration;
|
||||
use syncode_collab_store::{
|
||||
PostgresCollaboration, PostgresInbox, PostgresOutbox, connect, migrate,
|
||||
};
|
||||
use tower_http::cors::{AllowOrigin, CorsLayer};
|
||||
|
||||
use crate::checks::ControlChecks;
|
||||
use crate::config::Config;
|
||||
use crate::http::router;
|
||||
use crate::identity::IdentityClientAccess;
|
||||
use crate::outbox_worker::{OutboxWorker, OutboxWorkerConfig};
|
||||
use crate::repository::RepositoryClientAccess;
|
||||
use crate::webhook_worker::WebhookWorker;
|
||||
|
||||
pub async fn run(config: Config) -> Result<(), RuntimeError> {
|
||||
if config.event_secret.is_empty() {
|
||||
return Err(RuntimeError::Configuration(
|
||||
"SYNCODE_COLLAB_EVENT_SECRET must not be empty".to_owned(),
|
||||
));
|
||||
}
|
||||
if config.identity_shared_secret.is_empty() {
|
||||
return Err(RuntimeError::Configuration(
|
||||
"SYNCODE_IDENTITY_SHARED_SECRET must not be empty".to_owned(),
|
||||
));
|
||||
}
|
||||
let webhook_key = parse_webhook_key(&config.webhook_key)?;
|
||||
let pool = connect(&config.database_url, config.database_max_connections)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Store(error.to_string()))?;
|
||||
migrate(&pool)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Store(error.to_string()))?;
|
||||
|
||||
let http = tokio::net::TcpListener::bind(&config.listen_http)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Listen(config.listen_http.clone(), error.to_string()))?;
|
||||
let grpc = config
|
||||
.listen_grpc
|
||||
.parse()
|
||||
.map_err(|_| RuntimeError::Address(config.listen_grpc.clone()))?;
|
||||
|
||||
let identity = Arc::new(
|
||||
IdentityClientAccess::connect(config.identity_grpc_url, config.identity_shared_secret)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Identity(error.to_string()))?,
|
||||
);
|
||||
let repository = Arc::new(
|
||||
RepositoryClientAccess::connect(config.repository_grpc_url)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Repository(error.to_string()))?,
|
||||
);
|
||||
let checks = Arc::new(
|
||||
ControlChecks::connect(config.control_grpc_url)
|
||||
.await
|
||||
.map_err(|error| RuntimeError::Control(error.to_string()))?,
|
||||
);
|
||||
let store = PostgresCollaboration::with_webhook_key(pool.clone(), webhook_key);
|
||||
let collaboration = Arc::new(Collaboration::with_delivery(
|
||||
Arc::new(store.clone()),
|
||||
identity,
|
||||
repository,
|
||||
checks,
|
||||
));
|
||||
let inbox = Arc::new(PostgresInbox::new(pool.clone()));
|
||||
let outbox_worker = OutboxWorker::new(
|
||||
PostgresOutbox::new(pool.clone()),
|
||||
Arc::new(PostgresInbox::new(pool)),
|
||||
OutboxWorkerConfig {
|
||||
source_node_id: config.source_node_id,
|
||||
control_url: config.control_event_url,
|
||||
secret: config.event_secret.clone(),
|
||||
interval: Duration::from_secs(config.outbox_interval_seconds),
|
||||
batch: config.outbox_batch,
|
||||
lease: config.outbox_lease_seconds,
|
||||
},
|
||||
)
|
||||
.map_err(RuntimeError::Configuration)?
|
||||
.run();
|
||||
let mut http_router = router(
|
||||
inbox,
|
||||
Arc::clone(&collaboration),
|
||||
config.event_secret,
|
||||
config.session_cookie_name,
|
||||
);
|
||||
if !config.cors_origins.is_empty() {
|
||||
let allowed = config.cors_origins;
|
||||
http_router = http_router.layer(
|
||||
CorsLayer::new()
|
||||
.allow_credentials(true)
|
||||
.allow_methods([
|
||||
axum::http::Method::GET,
|
||||
axum::http::Method::POST,
|
||||
axum::http::Method::OPTIONS,
|
||||
])
|
||||
.allow_headers([axum::http::header::CONTENT_TYPE])
|
||||
.allow_origin(AllowOrigin::predicate(move |origin, _| {
|
||||
origin
|
||||
.to_str()
|
||||
.is_ok_and(|origin| allowed.iter().any(|value| value == origin))
|
||||
})),
|
||||
);
|
||||
}
|
||||
let serve_http = axum::serve(http, http_router);
|
||||
let serve_grpc = tonic::transport::Server::builder()
|
||||
.add_service(CollabService::new(collaboration).into_server())
|
||||
.serve(grpc);
|
||||
let webhook_worker = WebhookWorker::new(
|
||||
store,
|
||||
Duration::from_secs(config.webhook_interval_seconds),
|
||||
config.webhook_batch,
|
||||
)
|
||||
.map_err(RuntimeError::Configuration)?
|
||||
.run();
|
||||
|
||||
tokio::select! {
|
||||
result = serve_http => result.map_err(|error| RuntimeError::Serve(error.to_string())),
|
||||
result = serve_grpc => result.map_err(|error| RuntimeError::Serve(error.to_string())),
|
||||
result = webhook_worker => result.map_err(RuntimeError::Webhook),
|
||||
result = outbox_worker => result.map_err(RuntimeError::Outbox),
|
||||
_ = tokio::signal::ctrl_c() => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_webhook_key(value: &str) -> Result<[u8; 32], RuntimeError> {
|
||||
let decoded = hex::decode(value).map_err(|_| {
|
||||
RuntimeError::Configuration(
|
||||
"SYNCODE_COLLAB_WEBHOOK_KEY must be 64 hexadecimal characters".to_owned(),
|
||||
)
|
||||
})?;
|
||||
decoded.try_into().map_err(|_| {
|
||||
RuntimeError::Configuration(
|
||||
"SYNCODE_COLLAB_WEBHOOK_KEY must be 64 hexadecimal characters".to_owned(),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum RuntimeError {
|
||||
#[error("invalid collaboration configuration: {0}")]
|
||||
Configuration(String),
|
||||
#[error("{0} is not a valid socket address")]
|
||||
Address(String),
|
||||
#[error("cannot listen on {0}: {1}")]
|
||||
Listen(String, String),
|
||||
#[error("collaboration store failed: {0}")]
|
||||
Store(String),
|
||||
#[error("identity integration failed: {0}")]
|
||||
Identity(String),
|
||||
#[error("repository integration failed: {0}")]
|
||||
Repository(String),
|
||||
#[error("control integration failed: {0}")]
|
||||
Control(String),
|
||||
#[error("webhook delivery failed: {0}")]
|
||||
Webhook(String),
|
||||
#[error("event outbox failed: {0}")]
|
||||
Outbox(String),
|
||||
#[error("serving failed: {0}")]
|
||||
Serve(String),
|
||||
}
|
||||
@@ -1,79 +1,79 @@
|
||||
use sqlx::PgPool;
|
||||
use syncode_collab_application::StoreError;
|
||||
use syncode_collab_model::{PullRequest, PullRequestState};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::delivery_pull_requests::{COLUMNS, PullRequestRow};
|
||||
use crate::postgres::{outbox, unavailable};
|
||||
|
||||
pub async fn change(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
pull_request_id: Uuid,
|
||||
state: PullRequestState,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let target = match state {
|
||||
PullRequestState::Draft => "draft",
|
||||
PullRequestState::Open => "open",
|
||||
PullRequestState::Closed => "closed",
|
||||
PullRequestState::Merged => {
|
||||
return Err(StoreError::Conflict(
|
||||
"merged state can only be set by merge".to_owned(),
|
||||
));
|
||||
}
|
||||
};
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let query = format!(
|
||||
"UPDATE pull_request SET state=$2, updated_at=now() WHERE id=$1 AND \
|
||||
((state='draft' AND $2 IN ('open','closed')) OR \
|
||||
(state='open' AND $2='closed') OR (state='closed' AND $2='open')) \
|
||||
RETURNING {COLUMNS}"
|
||||
);
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(pull_request_id)
|
||||
.bind(target)
|
||||
.fetch_optional(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or_else(|| StoreError::Conflict("invalid pull request state transition".to_owned()))?;
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.state_changed",
|
||||
pull_request_id,
|
||||
serde_json::json!({"pull_request_id": pull_request_id, "state": target, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
|
||||
pub async fn synchronize(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
pull_request_id: Uuid,
|
||||
base_oid: &str,
|
||||
head_oid: &str,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let query = format!(
|
||||
"UPDATE pull_request SET base_oid=$2, head_oid=$3, updated_at=now() \
|
||||
WHERE id=$1 AND state IN ('draft','open','closed') RETURNING {COLUMNS}"
|
||||
);
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(pull_request_id)
|
||||
.bind(base_oid)
|
||||
.bind(head_oid)
|
||||
.fetch_optional(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or_else(|| StoreError::Conflict("pull request cannot be synchronized".to_owned()))?;
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.synchronized",
|
||||
pull_request_id,
|
||||
serde_json::json!({"pull_request_id": pull_request_id, "base_oid": base_oid, "head_oid": head_oid, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
use sqlx::PgPool;
|
||||
use syncode_collab_application::StoreError;
|
||||
use syncode_collab_model::{PullRequest, PullRequestState};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::delivery_pull_requests::{COLUMNS, PullRequestRow, event_payload};
|
||||
use crate::postgres::{outbox, unavailable};
|
||||
|
||||
pub async fn change(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
pull_request_id: Uuid,
|
||||
state: PullRequestState,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let target = match state {
|
||||
PullRequestState::Draft => "draft",
|
||||
PullRequestState::Open => "open",
|
||||
PullRequestState::Closed => "closed",
|
||||
PullRequestState::Merged => {
|
||||
return Err(StoreError::Conflict(
|
||||
"merged state can only be set by merge".to_owned(),
|
||||
));
|
||||
}
|
||||
};
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let query = format!(
|
||||
"UPDATE pull_request SET state=$2, updated_at=now() WHERE id=$1 AND \
|
||||
((state='draft' AND $2 IN ('open','closed')) OR \
|
||||
(state='open' AND $2='closed') OR (state='closed' AND $2='open')) \
|
||||
RETURNING {COLUMNS}"
|
||||
);
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(pull_request_id)
|
||||
.bind(target)
|
||||
.fetch_optional(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or_else(|| StoreError::Conflict("invalid pull request state transition".to_owned()))?;
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.state_changed",
|
||||
pull_request_id,
|
||||
event_payload(&row, principal_id),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
|
||||
pub async fn synchronize(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
pull_request_id: Uuid,
|
||||
base_oid: &str,
|
||||
head_oid: &str,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let query = format!(
|
||||
"UPDATE pull_request SET base_oid=$2, head_oid=$3, updated_at=now() \
|
||||
WHERE id=$1 AND state IN ('draft','open','closed') RETURNING {COLUMNS}"
|
||||
);
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(pull_request_id)
|
||||
.bind(base_oid)
|
||||
.bind(head_oid)
|
||||
.fetch_optional(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or_else(|| StoreError::Conflict("pull request cannot be synchronized".to_owned()))?;
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.synchronized",
|
||||
pull_request_id,
|
||||
event_payload(&row, principal_id),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
@@ -1,233 +1,249 @@
|
||||
use sqlx::{FromRow, PgPool};
|
||||
use syncode_collab_application::{CreatePullRequest, LinkIssue, StoreError};
|
||||
use syncode_collab_model::{PullRequest, PullRequestState};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::postgres::{outbox, rejected, unavailable};
|
||||
|
||||
#[derive(FromRow)]
|
||||
pub(crate) struct PullRequestRow {
|
||||
id: Uuid,
|
||||
repository_id: Uuid,
|
||||
number: i64,
|
||||
title: String,
|
||||
description: String,
|
||||
state: String,
|
||||
base_repository_id: Uuid,
|
||||
base_branch: String,
|
||||
base_oid: String,
|
||||
head_repository_id: Uuid,
|
||||
head_branch: String,
|
||||
head_oid: String,
|
||||
author_principal_id: Uuid,
|
||||
merged_oid: Option<String>,
|
||||
created_at: chrono::DateTime<chrono::Utc>,
|
||||
updated_at: chrono::DateTime<chrono::Utc>,
|
||||
}
|
||||
|
||||
impl TryFrom<PullRequestRow> for PullRequest {
|
||||
type Error = StoreError;
|
||||
|
||||
fn try_from(row: PullRequestRow) -> Result<Self, Self::Error> {
|
||||
let state = match row.state.as_str() {
|
||||
"draft" => PullRequestState::Draft,
|
||||
"open" => PullRequestState::Open,
|
||||
"closed" => PullRequestState::Closed,
|
||||
"merged" => PullRequestState::Merged,
|
||||
value => {
|
||||
return Err(StoreError::Rejected(format!(
|
||||
"unknown pull request state {value}"
|
||||
)));
|
||||
}
|
||||
};
|
||||
Ok(Self {
|
||||
id: row.id,
|
||||
repository_id: row.repository_id,
|
||||
number: row.number,
|
||||
title: row.title,
|
||||
description: row.description,
|
||||
state,
|
||||
base_repository_id: row.base_repository_id,
|
||||
base_branch: row.base_branch,
|
||||
base_oid: row.base_oid,
|
||||
head_repository_id: row.head_repository_id,
|
||||
head_branch: row.head_branch,
|
||||
head_oid: row.head_oid,
|
||||
author_principal_id: row.author_principal_id,
|
||||
merged_oid: row.merged_oid,
|
||||
created_at: row.created_at,
|
||||
updated_at: row.updated_at,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) const COLUMNS: &str = "id, repository_id, number, title, description, state, \
|
||||
base_repository_id, base_branch, base_oid, head_repository_id, head_branch, head_oid, \
|
||||
author_principal_id, merged_oid, created_at, updated_at";
|
||||
|
||||
pub async fn create(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
command: &CreatePullRequest,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let (number,): (i64,) = sqlx::query_as(
|
||||
"INSERT INTO repository_pull_request_counter (repository_id, next_number) VALUES ($1, 2) \
|
||||
ON CONFLICT (repository_id) DO UPDATE SET next_number = \
|
||||
repository_pull_request_counter.next_number + 1 RETURNING next_number - 1",
|
||||
)
|
||||
.bind(command.repository_id)
|
||||
.fetch_one(&mut *transaction)
|
||||
.await
|
||||
.map_err(rejected)?;
|
||||
let id = Uuid::new_v4();
|
||||
let query = format!(
|
||||
"INSERT INTO pull_request ({COLUMNS}) VALUES \
|
||||
($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,NULL,now(),now()) RETURNING {COLUMNS}"
|
||||
);
|
||||
let state = if command.draft { "draft" } else { "open" };
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(id)
|
||||
.bind(command.repository_id)
|
||||
.bind(number)
|
||||
.bind(&command.title)
|
||||
.bind(&command.description)
|
||||
.bind(state)
|
||||
.bind(command.base_repository_id)
|
||||
.bind(&command.base_branch)
|
||||
.bind(&command.base_oid)
|
||||
.bind(command.head_repository_id)
|
||||
.bind(&command.head_branch)
|
||||
.bind(&command.head_oid)
|
||||
.bind(principal_id)
|
||||
.fetch_one(&mut *transaction)
|
||||
.await
|
||||
.map_err(rejected)?;
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.created",
|
||||
id,
|
||||
serde_json::json!({"pull_request_id": id, "repository_id": command.repository_id, "number": number, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
|
||||
pub async fn list(pool: &PgPool, repository_id: Uuid) -> Result<Vec<PullRequest>, StoreError> {
|
||||
let query =
|
||||
format!("SELECT {COLUMNS} FROM pull_request WHERE repository_id = $1 ORDER BY number DESC");
|
||||
sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(repository_id)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.into_iter()
|
||||
.map(TryInto::try_into)
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn get(
|
||||
pool: &PgPool,
|
||||
repository_id: Uuid,
|
||||
number: i64,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let query =
|
||||
format!("SELECT {COLUMNS} FROM pull_request WHERE repository_id = $1 AND number = $2");
|
||||
sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(repository_id)
|
||||
.bind(number)
|
||||
.fetch_optional(pool)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or(StoreError::NotFound)?
|
||||
.try_into()
|
||||
}
|
||||
|
||||
pub async fn link_issue(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
command: &LinkIssue,
|
||||
) -> Result<(), StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let linked = sqlx::query(
|
||||
"INSERT INTO pull_request_issue (pull_request_id, issue_id, closes) \
|
||||
SELECT $1, $2, $3 FROM pull_request pr JOIN issue i ON i.id = $2 \
|
||||
JOIN project p ON p.id = i.project_id \
|
||||
JOIN workspace_access a ON a.workspace_id = p.workspace_id \
|
||||
JOIN workspace_repository r ON r.workspace_id = p.workspace_id \
|
||||
AND r.repository_id IN (pr.base_repository_id, pr.head_repository_id) \
|
||||
WHERE pr.id = $1 AND a.principal_id = $4 AND 'read' = ANY(a.capabilities) \
|
||||
ON CONFLICT (pull_request_id, issue_id) DO UPDATE SET closes = EXCLUDED.closes",
|
||||
)
|
||||
.bind(command.pull_request_id)
|
||||
.bind(command.issue_id)
|
||||
.bind(command.closes)
|
||||
.bind(principal_id)
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.map_err(rejected)?;
|
||||
if linked.rows_affected() == 0 {
|
||||
return Err(StoreError::NotFound);
|
||||
}
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.issue_linked",
|
||||
command.pull_request_id,
|
||||
serde_json::json!({"pull_request_id": command.pull_request_id, "issue_id": command.issue_id, "closes": command.closes, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn mark_merged(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
pull_request_id: Uuid,
|
||||
merged_oid: &str,
|
||||
close_issues: bool,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let query = format!(
|
||||
"UPDATE pull_request SET state = 'merged', merged_oid = $2, updated_at = now() \
|
||||
WHERE id = $1 AND state = 'open' RETURNING {COLUMNS}"
|
||||
);
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(pull_request_id)
|
||||
.bind(merged_oid)
|
||||
.fetch_optional(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or_else(|| StoreError::Conflict("pull request is not open".to_owned()))?;
|
||||
if close_issues {
|
||||
let issue_ids: Vec<(Uuid,)> = sqlx::query_as(
|
||||
"UPDATE issue i SET status_id = p.completion_status_id, version = i.version + 1, \
|
||||
updated_at = now() FROM project p, pull_request_issue l \
|
||||
WHERE l.pull_request_id = $1 AND l.closes AND i.id = l.issue_id \
|
||||
AND p.id = i.project_id RETURNING i.id",
|
||||
)
|
||||
.bind(pull_request_id)
|
||||
.fetch_all(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?;
|
||||
for (issue_id,) in issue_ids {
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.issue.closed",
|
||||
issue_id,
|
||||
serde_json::json!({"issue_id": issue_id, "pull_request_id": pull_request_id, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.merged",
|
||||
pull_request_id,
|
||||
serde_json::json!({"pull_request_id": pull_request_id, "merged_oid": merged_oid, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
use sqlx::{FromRow, PgPool};
|
||||
use syncode_collab_application::{CreatePullRequest, LinkIssue, StoreError};
|
||||
use syncode_collab_model::{PullRequest, PullRequestState};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::postgres::{outbox, rejected, unavailable};
|
||||
|
||||
#[derive(FromRow)]
|
||||
pub(crate) struct PullRequestRow {
|
||||
id: Uuid,
|
||||
repository_id: Uuid,
|
||||
number: i64,
|
||||
title: String,
|
||||
description: String,
|
||||
state: String,
|
||||
base_repository_id: Uuid,
|
||||
base_branch: String,
|
||||
base_oid: String,
|
||||
head_repository_id: Uuid,
|
||||
head_branch: String,
|
||||
head_oid: String,
|
||||
author_principal_id: Uuid,
|
||||
merged_oid: Option<String>,
|
||||
created_at: chrono::DateTime<chrono::Utc>,
|
||||
updated_at: chrono::DateTime<chrono::Utc>,
|
||||
}
|
||||
|
||||
pub(crate) fn event_payload(row: &PullRequestRow, actor_id: Uuid) -> serde_json::Value {
|
||||
serde_json::json!({
|
||||
"pull_request_id": row.id,
|
||||
"repository_id": row.repository_id,
|
||||
"number": row.number,
|
||||
"base_repository_id": row.base_repository_id,
|
||||
"base_branch": row.base_branch,
|
||||
"base_oid": row.base_oid,
|
||||
"head_repository_id": row.head_repository_id,
|
||||
"head_branch": row.head_branch,
|
||||
"head_oid": row.head_oid,
|
||||
"actor_id": actor_id,
|
||||
"state": row.state,
|
||||
})
|
||||
}
|
||||
|
||||
impl TryFrom<PullRequestRow> for PullRequest {
|
||||
type Error = StoreError;
|
||||
|
||||
fn try_from(row: PullRequestRow) -> Result<Self, Self::Error> {
|
||||
let state = match row.state.as_str() {
|
||||
"draft" => PullRequestState::Draft,
|
||||
"open" => PullRequestState::Open,
|
||||
"closed" => PullRequestState::Closed,
|
||||
"merged" => PullRequestState::Merged,
|
||||
value => {
|
||||
return Err(StoreError::Rejected(format!(
|
||||
"unknown pull request state {value}"
|
||||
)));
|
||||
}
|
||||
};
|
||||
Ok(Self {
|
||||
id: row.id,
|
||||
repository_id: row.repository_id,
|
||||
number: row.number,
|
||||
title: row.title,
|
||||
description: row.description,
|
||||
state,
|
||||
base_repository_id: row.base_repository_id,
|
||||
base_branch: row.base_branch,
|
||||
base_oid: row.base_oid,
|
||||
head_repository_id: row.head_repository_id,
|
||||
head_branch: row.head_branch,
|
||||
head_oid: row.head_oid,
|
||||
author_principal_id: row.author_principal_id,
|
||||
merged_oid: row.merged_oid,
|
||||
created_at: row.created_at,
|
||||
updated_at: row.updated_at,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) const COLUMNS: &str = "id, repository_id, number, title, description, state, \
|
||||
base_repository_id, base_branch, base_oid, head_repository_id, head_branch, head_oid, \
|
||||
author_principal_id, merged_oid, created_at, updated_at";
|
||||
|
||||
pub async fn create(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
command: &CreatePullRequest,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let (number,): (i64,) = sqlx::query_as(
|
||||
"INSERT INTO repository_pull_request_counter (repository_id, next_number) VALUES ($1, 2) \
|
||||
ON CONFLICT (repository_id) DO UPDATE SET next_number = \
|
||||
repository_pull_request_counter.next_number + 1 RETURNING next_number - 1",
|
||||
)
|
||||
.bind(command.repository_id)
|
||||
.fetch_one(&mut *transaction)
|
||||
.await
|
||||
.map_err(rejected)?;
|
||||
let id = Uuid::new_v4();
|
||||
let query = format!(
|
||||
"INSERT INTO pull_request ({COLUMNS}) VALUES \
|
||||
($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,NULL,now(),now()) RETURNING {COLUMNS}"
|
||||
);
|
||||
let state = if command.draft { "draft" } else { "open" };
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(id)
|
||||
.bind(command.repository_id)
|
||||
.bind(number)
|
||||
.bind(&command.title)
|
||||
.bind(&command.description)
|
||||
.bind(state)
|
||||
.bind(command.base_repository_id)
|
||||
.bind(&command.base_branch)
|
||||
.bind(&command.base_oid)
|
||||
.bind(command.head_repository_id)
|
||||
.bind(&command.head_branch)
|
||||
.bind(&command.head_oid)
|
||||
.bind(principal_id)
|
||||
.fetch_one(&mut *transaction)
|
||||
.await
|
||||
.map_err(rejected)?;
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.created",
|
||||
id,
|
||||
event_payload(&row, principal_id),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
|
||||
pub async fn list(pool: &PgPool, repository_id: Uuid) -> Result<Vec<PullRequest>, StoreError> {
|
||||
let query =
|
||||
format!("SELECT {COLUMNS} FROM pull_request WHERE repository_id = $1 ORDER BY number DESC");
|
||||
sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(repository_id)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.into_iter()
|
||||
.map(TryInto::try_into)
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn get(
|
||||
pool: &PgPool,
|
||||
repository_id: Uuid,
|
||||
number: i64,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let query =
|
||||
format!("SELECT {COLUMNS} FROM pull_request WHERE repository_id = $1 AND number = $2");
|
||||
sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(repository_id)
|
||||
.bind(number)
|
||||
.fetch_optional(pool)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or(StoreError::NotFound)?
|
||||
.try_into()
|
||||
}
|
||||
|
||||
pub async fn link_issue(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
command: &LinkIssue,
|
||||
) -> Result<(), StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let linked = sqlx::query(
|
||||
"INSERT INTO pull_request_issue (pull_request_id, issue_id, closes) \
|
||||
SELECT $1, $2, $3 FROM pull_request pr JOIN issue i ON i.id = $2 \
|
||||
JOIN project p ON p.id = i.project_id \
|
||||
JOIN workspace_access a ON a.workspace_id = p.workspace_id \
|
||||
JOIN workspace_repository r ON r.workspace_id = p.workspace_id \
|
||||
AND r.repository_id IN (pr.base_repository_id, pr.head_repository_id) \
|
||||
WHERE pr.id = $1 AND a.principal_id = $4 AND 'read' = ANY(a.capabilities) \
|
||||
ON CONFLICT (pull_request_id, issue_id) DO UPDATE SET closes = EXCLUDED.closes",
|
||||
)
|
||||
.bind(command.pull_request_id)
|
||||
.bind(command.issue_id)
|
||||
.bind(command.closes)
|
||||
.bind(principal_id)
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.map_err(rejected)?;
|
||||
if linked.rows_affected() == 0 {
|
||||
return Err(StoreError::NotFound);
|
||||
}
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.issue_linked",
|
||||
command.pull_request_id,
|
||||
serde_json::json!({"pull_request_id": command.pull_request_id, "issue_id": command.issue_id, "closes": command.closes, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn mark_merged(
|
||||
pool: &PgPool,
|
||||
principal_id: Uuid,
|
||||
pull_request_id: Uuid,
|
||||
merged_oid: &str,
|
||||
close_issues: bool,
|
||||
) -> Result<PullRequest, StoreError> {
|
||||
let mut transaction = pool.begin().await.map_err(unavailable)?;
|
||||
let query = format!(
|
||||
"UPDATE pull_request SET state = 'merged', merged_oid = $2, updated_at = now() \
|
||||
WHERE id = $1 AND state = 'open' RETURNING {COLUMNS}"
|
||||
);
|
||||
let row = sqlx::query_as::<_, PullRequestRow>(&query)
|
||||
.bind(pull_request_id)
|
||||
.bind(merged_oid)
|
||||
.fetch_optional(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?
|
||||
.ok_or_else(|| StoreError::Conflict("pull request is not open".to_owned()))?;
|
||||
if close_issues {
|
||||
let issue_ids: Vec<(Uuid,)> = sqlx::query_as(
|
||||
"UPDATE issue i SET status_id = p.completion_status_id, version = i.version + 1, \
|
||||
updated_at = now() FROM project p, pull_request_issue l \
|
||||
WHERE l.pull_request_id = $1 AND l.closes AND i.id = l.issue_id \
|
||||
AND p.id = i.project_id RETURNING i.id",
|
||||
)
|
||||
.bind(pull_request_id)
|
||||
.fetch_all(&mut *transaction)
|
||||
.await
|
||||
.map_err(unavailable)?;
|
||||
for (issue_id,) in issue_ids {
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.issue.closed",
|
||||
issue_id,
|
||||
serde_json::json!({"issue_id": issue_id, "pull_request_id": pull_request_id, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
outbox(
|
||||
&mut transaction,
|
||||
"collaboration.pull_request.merged",
|
||||
pull_request_id,
|
||||
serde_json::json!({"pull_request_id": pull_request_id, "merged_oid": merged_oid, "actor_id": principal_id}),
|
||||
)
|
||||
.await?;
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
row.try_into()
|
||||
}
|
||||
@@ -1,0 +1,202 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use hmac::{Hmac, KeyInit, Mac};
|
||||
use sha2::Sha256;
|
||||
use syncode_collab_application::{
|
||||
Acceptance, EventInbox, EventPublisher, OutboxMessage, OutboxQueue, PublishError, accept,
|
||||
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,
|
||||
inbox: Arc<dyn EventInbox>,
|
||||
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 {
|
||||
inbox,
|
||||
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 {
|
||||
inbox: Arc<dyn EventInbox>,
|
||||
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)?;
|
||||
match accept(self.inbox.as_ref(), &envelope).await {
|
||||
Ok(Acceptance::Applied | Acceptance::AlreadyApplied) => {}
|
||||
Err(error) => return Err(PublishError::Unreachable(error.to_string())),
|
||||
}
|
||||
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"}),
|
||||
)));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user