feat: accept plane events through durable inbox #1
+114
-21
File diff suppressed because it is too large
Load Diff
@@ -1,57 +1,61 @@
|
||||
[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-axum = "7.2.1"
|
||||
axum = "0.8.9"
|
||||
clap = { version = "4.6.4", features = ["derive", "env"] }
|
||||
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"
|
||||
|
||||
[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-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"
|
||||
serde_json = "1.0.149"
|
||||
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"
|
||||
|
||||
[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,43 +1,44 @@
|
||||
# syncode-collab
|
||||
|
||||
The SynCode collaboration plane. It is the authority for Workspaces, Projects,
|
||||
Issues, pull requests, review, Documents, Versions, webhooks, the activity feed
|
||||
and the Inbox.
|
||||
|
||||
Collaboration keeps stable `repository_id` and `principal_id` values but never
|
||||
joins another plane's database. Facts about refs, tags, principals and checks
|
||||
arrive as versioned events through a deduplicating inbox; decisions and state
|
||||
changes leave as commands over typed ports. There is no fallback path, no
|
||||
dual-write and no read of a foreign store.
|
||||
|
||||
## Layout
|
||||
|
||||
| Crate | Role |
|
||||
| --- | --- |
|
||||
| `collab-model` | domain types; free of transport, storage and runtime |
|
||||
| `collab-application` | use cases and ports; free of transport and storage |
|
||||
| `collab-store` | PostgreSQL state, migrations, inbox and outbox |
|
||||
| `collab-api-graphql` | typed GraphQL surface for the front |
|
||||
| `collab-api-grpc` | versioned gRPC surface for `syn` |
|
||||
|
||||
## Running
|
||||
|
||||
```sh
|
||||
export SYNCODE_COLLAB_DATABASE_URL=postgres://collab:collab@localhost:5432/collab
|
||||
cargo run --bin syncode-collab
|
||||
```
|
||||
|
||||
Migrations are applied on start, so an empty database is a valid starting point.
|
||||
gRPC listens on `8300` and HTTP on `8301`.
|
||||
|
||||
## Contracts
|
||||
|
||||
`crates/collab-api-graphql/schema.graphql` is the committed GraphQL contract.
|
||||
Regenerate it after changing the schema and commit the result:
|
||||
|
||||
```sh
|
||||
cargo run --bin collab_schema > crates/collab-api-graphql/schema.graphql
|
||||
```
|
||||
|
||||
The gRPC contract lives in `crates/collab-api-grpc/proto/collab.proto` and is
|
||||
generated at build time. CI fails when either contract drifts from its source.
|
||||
# syncode-collab
|
||||
|
||||
The SynCode collaboration plane. It is the authority for Workspaces, Projects,
|
||||
Issues, pull requests, review, Documents, Versions, webhooks, the activity feed
|
||||
and the Inbox.
|
||||
|
||||
Collaboration keeps stable `repository_id` and `principal_id` values but never
|
||||
joins another plane's database. Facts about refs, tags, principals and checks
|
||||
arrive as versioned events through a deduplicating inbox; decisions and state
|
||||
changes leave as commands over typed ports. There is no fallback path, no
|
||||
dual-write and no read of a foreign store.
|
||||
|
||||
## Layout
|
||||
|
||||
| Crate | Role |
|
||||
| --- | --- |
|
||||
| `collab-model` | domain types; free of transport, storage and runtime |
|
||||
| `collab-application` | use cases and ports; free of transport and storage |
|
||||
| `collab-store` | PostgreSQL state, migrations, inbox and outbox |
|
||||
| `collab-api-graphql` | typed GraphQL surface for the front |
|
||||
| `collab-api-grpc` | versioned gRPC surface for `syn` |
|
||||
|
||||
## Running
|
||||
|
||||
```sh
|
||||
export SYNCODE_COLLAB_DATABASE_URL=postgres://collab:collab@localhost:5432/collab
|
||||
export SYNCODE_COLLAB_EVENT_SECRET=replace-with-a-shared-secret
|
||||
cargo run --bin syncode-collab
|
||||
```
|
||||
|
||||
Migrations are applied on start, so an empty database is a valid starting point.
|
||||
gRPC listens on `8300` and HTTP on `8301`.
|
||||
|
||||
## Contracts
|
||||
|
||||
`crates/collab-api-graphql/schema.graphql` is the committed GraphQL contract.
|
||||
Regenerate it after changing the schema and commit the result:
|
||||
|
||||
```sh
|
||||
cargo run --bin collab_schema > crates/collab-api-graphql/schema.graphql
|
||||
```
|
||||
|
||||
The gRPC contract lives in `crates/collab-api-grpc/proto/collab.proto` and is
|
||||
generated at build time. CI fails when either contract drifts from its source.
|
||||
@@ -1,29 +1,32 @@
|
||||
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,
|
||||
}
|
||||
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,
|
||||
}
|
||||
+80
-12
@@ -1,29 +1,97 @@
|
||||
use async_graphql_axum::GraphQL;
|
||||
use axum::Router;
|
||||
use axum::routing::{get, post};
|
||||
use std::sync::Arc;
|
||||
use syncode_collab_application::{Acceptance, EventInbox, InboxError, accept};
|
||||
use syncode_collab_model::Envelope;
|
||||
|
||||
pub fn router(inbox: Arc<dyn EventInbox>) -> Router {
|
||||
Router::new()
|
||||
.route("/healthz", get(|| async { "ok" }))
|
||||
.route("/events", post(receive_event))
|
||||
.route_service(
|
||||
"/graphql",
|
||||
GraphQL::new(syncode_collab_api_graphql::build()),
|
||||
)
|
||||
.with_state(inbox)
|
||||
}
|
||||
|
||||
async fn receive_event(
|
||||
axum::extract::State(inbox): axum::extract::State<Arc<dyn EventInbox>>,
|
||||
axum::Json(envelope): axum::Json<Envelope>,
|
||||
) -> axum::http::StatusCode {
|
||||
match accept(inbox.as_ref(), &envelope).await {
|
||||
Ok(Acceptance::Applied) => axum::http::StatusCode::NO_CONTENT,
|
||||
Ok(Acceptance::AlreadyApplied) => axum::http::StatusCode::OK,
|
||||
Err(InboxError::Store(_)) => axum::http::StatusCode::SERVICE_UNAVAILABLE,
|
||||
Err(_) => axum::http::StatusCode::BAD_REQUEST,
|
||||
}
|
||||
}
|
||||
use async_graphql_axum::GraphQL;
|
||||
use axum::body::Bytes;
|
||||
use axum::http::{HeaderMap, StatusCode};
|
||||
use axum::routing::{get, post};
|
||||
use axum::{Router, extract::State};
|
||||
use hmac::{Hmac, KeyInit, Mac};
|
||||
use sha2::Sha256;
|
||||
use std::sync::Arc;
|
||||
use syncode_collab_application::{Acceptance, EventInbox, InboxError, StoreError, accept};
|
||||
use syncode_collab_model::Envelope;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct HttpState {
|
||||
inbox: Arc<dyn EventInbox>,
|
||||
event_secret: String,
|
||||
}
|
||||
|
||||
pub fn router(inbox: Arc<dyn EventInbox>, event_secret: String) -> Router {
|
||||
Router::new()
|
||||
.route("/healthz", get(|| async { "ok" }))
|
||||
.route("/events", post(receive_event))
|
||||
.route_service(
|
||||
"/graphql",
|
||||
GraphQL::new(syncode_collab_api_graphql::build()),
|
||||
)
|
||||
.with_state(HttpState {
|
||||
inbox,
|
||||
event_secret,
|
||||
})
|
||||
}
|
||||
|
||||
async fn receive_event(
|
||||
State(state): State<HttpState>,
|
||||
headers: HeaderMap,
|
||||
body: Bytes,
|
||||
) -> StatusCode {
|
||||
if !valid_signature(&headers, &body, &state.event_secret) {
|
||||
return StatusCode::UNAUTHORIZED;
|
||||
}
|
||||
let Ok(envelope) = serde_json::from_slice::<Envelope>(&body) else {
|
||||
return StatusCode::BAD_REQUEST;
|
||||
};
|
||||
if headers
|
||||
.get("x-syncode-event")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some(envelope.message_type.as_str())
|
||||
|| headers
|
||||
.get("x-syncode-delivery")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some(envelope.message_id.to_string().as_str())
|
||||
{
|
||||
return StatusCode::BAD_REQUEST;
|
||||
}
|
||||
match accept(state.inbox.as_ref(), &envelope).await {
|
||||
Ok(Acceptance::Applied) => StatusCode::NO_CONTENT,
|
||||
Ok(Acceptance::AlreadyApplied) => StatusCode::OK,
|
||||
Err(InboxError::Store(StoreError::Unavailable(_))) => StatusCode::SERVICE_UNAVAILABLE,
|
||||
Err(InboxError::Store(StoreError::Rejected(_))) => StatusCode::BAD_REQUEST,
|
||||
Err(_) => StatusCode::BAD_REQUEST,
|
||||
}
|
||||
}
|
||||
|
||||
fn valid_signature(headers: &HeaderMap, body: &[u8], secret: &str) -> bool {
|
||||
let Some(signature) = headers
|
||||
.get("x-syncode-signature")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
.and_then(|value| hex::decode(value).ok())
|
||||
else {
|
||||
return false;
|
||||
};
|
||||
let Ok(mut mac) = Hmac::<Sha256>::new_from_slice(secret.as_bytes()) else {
|
||||
return false;
|
||||
};
|
||||
mac.update(body);
|
||||
mac.verify_slice(&signature).is_ok()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn verifies_the_raw_event_body() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let body = br#"{"message_id":"00000000-0000-0000-0000-000000000000"}"#;
|
||||
let mut mac = Hmac::<Sha256>::new_from_slice(b"secret")?;
|
||||
mac.update(body);
|
||||
let mut headers = HeaderMap::new();
|
||||
headers.insert(
|
||||
"x-syncode-signature",
|
||||
hex::encode(mac.finalize().into_bytes()).parse()?,
|
||||
);
|
||||
|
||||
assert!(valid_signature(&headers, body, "secret"));
|
||||
assert!(!valid_signature(&headers, b"changed", "secret"));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
+8
-1
@@ -1,47 +1,54 @@
|
||||
use std::sync::Arc;
|
||||
use syncode_collab_api_grpc::CollabService;
|
||||
use syncode_collab_store::{PostgresInbox, connect, migrate};
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::http::router;
|
||||
|
||||
pub async fn run(config: Config) -> Result<(), RuntimeError> {
|
||||
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 inbox = Arc::new(PostgresInbox::new(pool));
|
||||
let serve_http = axum::serve(http, router(inbox));
|
||||
let serve_grpc = tonic::transport::Server::builder()
|
||||
.add_service(CollabService.into_server())
|
||||
.serve(grpc);
|
||||
|
||||
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())),
|
||||
_ = tokio::signal::ctrl_c() => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum RuntimeError {
|
||||
#[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("serving failed: {0}")]
|
||||
Serve(String),
|
||||
}
|
||||
use std::sync::Arc;
|
||||
use syncode_collab_api_grpc::CollabService;
|
||||
use syncode_collab_store::{PostgresInbox, connect, migrate};
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::http::router;
|
||||
|
||||
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(),
|
||||
));
|
||||
}
|
||||
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 inbox = Arc::new(PostgresInbox::new(pool));
|
||||
let serve_http = axum::serve(http, router(inbox, config.event_secret));
|
||||
let serve_grpc = tonic::transport::Server::builder()
|
||||
.add_service(CollabService.into_server())
|
||||
.serve(grpc);
|
||||
|
||||
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())),
|
||||
_ = tokio::signal::ctrl_c() => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
#[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("serving failed: {0}")]
|
||||
Serve(String),
|
||||
}
|
||||
@@ -1,53 +1,53 @@
|
||||
use syncode_collab_model::{Envelope, EnvelopeError};
|
||||
|
||||
use crate::ports::{EventInbox, StoreError};
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
pub enum Acceptance {
|
||||
Applied,
|
||||
AlreadyApplied,
|
||||
}
|
||||
|
||||
/// Accepts one foreign fact. A repeated message is acknowledged without being
|
||||
/// applied again, and a message older than the last seen position for its
|
||||
/// resource is refused rather than reordered.
|
||||
pub async fn accept(inbox: &dyn EventInbox, envelope: &Envelope) -> Result<Acceptance, InboxError> {
|
||||
envelope.validate()?;
|
||||
|
||||
if inbox.already_applied(envelope.message_id).await? {
|
||||
return Ok(Acceptance::AlreadyApplied);
|
||||
}
|
||||
|
||||
if let Some((sequence, term)) = inbox
|
||||
.last_position(&envelope.source_node_id, envelope.resource_id)
|
||||
.await?
|
||||
{
|
||||
if envelope.term < term {
|
||||
return Err(InboxError::StaleTerm {
|
||||
seen: term,
|
||||
received: envelope.term,
|
||||
});
|
||||
}
|
||||
if envelope.term == term && envelope.sequence <= sequence {
|
||||
return Err(InboxError::StaleSequence {
|
||||
seen: sequence,
|
||||
received: envelope.sequence,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
inbox.apply(envelope).await?;
|
||||
Ok(Acceptance::Applied)
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum InboxError {
|
||||
#[error(transparent)]
|
||||
Envelope(#[from] EnvelopeError),
|
||||
#[error(transparent)]
|
||||
Store(#[from] StoreError),
|
||||
#[error("term {received} is older than the last seen term {seen}")]
|
||||
StaleTerm { seen: i64, received: i64 },
|
||||
#[error("sequence {received} is not after the last seen sequence {seen}")]
|
||||
StaleSequence { seen: i64, received: i64 },
|
||||
}
|
||||
use syncode_collab_model::{Envelope, EnvelopeError};
|
||||
|
||||
use crate::ports::{EventInbox, StoreError};
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
pub enum Acceptance {
|
||||
Applied,
|
||||
AlreadyApplied,
|
||||
}
|
||||
|
||||
/// Accepts one foreign fact. A repeated message is acknowledged without being
|
||||
/// applied again, and a message older than the last seen position for its
|
||||
/// resource is refused rather than reordered.
|
||||
pub async fn accept(inbox: &dyn EventInbox, envelope: &Envelope) -> Result<Acceptance, InboxError> {
|
||||
envelope.validate()?;
|
||||
|
||||
if inbox.already_applied(envelope.message_id).await? {
|
||||
return Ok(Acceptance::AlreadyApplied);
|
||||
}
|
||||
|
||||
if let Some((sequence, term)) = inbox
|
||||
.last_position(&envelope.source_node_id, envelope.repository_id)
|
||||
.await?
|
||||
{
|
||||
if envelope.term < term {
|
||||
return Err(InboxError::StaleTerm {
|
||||
seen: term,
|
||||
received: envelope.term,
|
||||
});
|
||||
}
|
||||
if envelope.term == term && envelope.sequence <= sequence {
|
||||
return Err(InboxError::StaleSequence {
|
||||
seen: sequence,
|
||||
received: envelope.sequence,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
inbox.apply(envelope).await?;
|
||||
Ok(Acceptance::Applied)
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum InboxError {
|
||||
#[error(transparent)]
|
||||
Envelope(#[from] EnvelopeError),
|
||||
#[error(transparent)]
|
||||
Store(#[from] StoreError),
|
||||
#[error("term {received} is older than the last seen term {seen}")]
|
||||
StaleTerm { seen: i64, received: i64 },
|
||||
#[error("sequence {received} is not after the last seen sequence {seen}")]
|
||||
StaleSequence { seen: i64, received: i64 },
|
||||
}
|
||||
@@ -1,133 +1,179 @@
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::plane::Plane;
|
||||
|
||||
pub const PROTOCOL_VERSION: ProtocolVersion = ProtocolVersion(1);
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
|
||||
pub struct ProtocolVersion(pub u32);
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
|
||||
pub struct SchemaVersion(pub u32);
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct MessageType(String);
|
||||
|
||||
impl MessageType {
|
||||
pub fn new(value: impl Into<String>) -> Result<Self, EnvelopeError> {
|
||||
let value = value.into();
|
||||
if value.is_empty() {
|
||||
return Err(EnvelopeError::EmptyMessageType);
|
||||
}
|
||||
Ok(Self(value))
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
/// Envelope carried by every cross-plane fact. The shape is the one
|
||||
/// `syncode-repo` settled on in 0.5; collaboration reuses it rather than
|
||||
/// introducing a second format.
|
||||
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct Envelope {
|
||||
pub message_id: Uuid,
|
||||
pub protocol_version: ProtocolVersion,
|
||||
pub schema_version: SchemaVersion,
|
||||
pub source_node_id: String,
|
||||
pub source_plane: String,
|
||||
pub resource_id: Uuid,
|
||||
pub message_type: MessageType,
|
||||
pub sequence: i64,
|
||||
pub term: i64,
|
||||
pub payload: serde_json::Value,
|
||||
}
|
||||
|
||||
impl Envelope {
|
||||
pub fn validate(&self) -> Result<Plane, EnvelopeError> {
|
||||
if self.protocol_version != PROTOCOL_VERSION {
|
||||
return Err(EnvelopeError::UnsupportedProtocol(self.protocol_version));
|
||||
}
|
||||
if self.source_node_id.is_empty() {
|
||||
return Err(EnvelopeError::EmptySourceNode);
|
||||
}
|
||||
if self.sequence < 0 {
|
||||
return Err(EnvelopeError::NegativeSequence(self.sequence));
|
||||
}
|
||||
if self.term < 0 {
|
||||
return Err(EnvelopeError::NegativeTerm(self.term));
|
||||
}
|
||||
self.source_plane
|
||||
.parse()
|
||||
.map_err(|_| EnvelopeError::UnknownSourcePlane(self.source_plane.clone()))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum EnvelopeError {
|
||||
#[error("message type must not be empty")]
|
||||
EmptyMessageType,
|
||||
#[error("source node id must not be empty")]
|
||||
EmptySourceNode,
|
||||
#[error("sequence must not be negative: {0}")]
|
||||
NegativeSequence(i64),
|
||||
#[error("term must not be negative: {0}")]
|
||||
NegativeTerm(i64),
|
||||
#[error("unsupported protocol version: {}", .0.0)]
|
||||
UnsupportedProtocol(ProtocolVersion),
|
||||
#[error("unknown source plane: {0}")]
|
||||
UnknownSourcePlane(String),
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn envelope() -> Envelope {
|
||||
Envelope {
|
||||
message_id: Uuid::nil(),
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
schema_version: SchemaVersion(1),
|
||||
source_node_id: "node-1".to_owned(),
|
||||
source_plane: "repository".to_owned(),
|
||||
resource_id: Uuid::nil(),
|
||||
message_type: MessageType(String::from("repository.ref.updated")),
|
||||
sequence: 7,
|
||||
term: 3,
|
||||
payload: serde_json::Value::Null,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_a_repository_envelope() {
|
||||
assert!(matches!(envelope().validate(), Ok(Plane::Repository)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_a_future_protocol() {
|
||||
let mut subject = envelope();
|
||||
subject.protocol_version = ProtocolVersion(PROTOCOL_VERSION.0 + 1);
|
||||
assert!(subject.validate().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_an_unknown_source_plane() {
|
||||
let mut subject = envelope();
|
||||
subject.source_plane = "forge".to_owned();
|
||||
assert!(subject.validate().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_unknown_fields() {
|
||||
let json = r#"{"message_id":"00000000-0000-0000-0000-000000000000","extra":1}"#;
|
||||
assert!(serde_json::from_str::<Envelope>(json).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_an_empty_message_type() {
|
||||
assert!(MessageType::new("").is_err());
|
||||
}
|
||||
}
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::plane::Plane;
|
||||
|
||||
pub const PROTOCOL_VERSION: ProtocolVersion = ProtocolVersion(1);
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
|
||||
#[serde(try_from = "String", into = "String")]
|
||||
pub struct ProtocolVersion(pub u32);
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
|
||||
pub struct SchemaVersion(pub u32);
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct MessageType(String);
|
||||
|
||||
impl MessageType {
|
||||
pub fn new(value: impl Into<String>) -> Result<Self, EnvelopeError> {
|
||||
let value = value.into();
|
||||
if value.is_empty() {
|
||||
return Err(EnvelopeError::EmptyMessageType);
|
||||
}
|
||||
Ok(Self(value))
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> &str {
|
||||
&self.0
|
||||
}
|
||||
|
||||
pub fn source_plane(&self) -> Result<Plane, EnvelopeError> {
|
||||
let source = self
|
||||
.0
|
||||
.split_once('.')
|
||||
.map_or(self.0.as_str(), |value| value.0);
|
||||
if source == "check" {
|
||||
return Ok(Plane::Control);
|
||||
}
|
||||
source
|
||||
.parse()
|
||||
.map_err(|_| EnvelopeError::UnknownSourcePlane(source.to_owned()))
|
||||
}
|
||||
}
|
||||
|
||||
/// Envelope carried by every cross-plane fact. The shape is the one
|
||||
/// `syncode-repo` settled on in 0.5; collaboration reuses it rather than
|
||||
/// introducing a second format.
|
||||
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct Envelope {
|
||||
pub message_id: Uuid,
|
||||
pub protocol_version: ProtocolVersion,
|
||||
pub schema_version: SchemaVersion,
|
||||
pub source_node_id: String,
|
||||
pub repository_id: Uuid,
|
||||
pub message_type: MessageType,
|
||||
pub sequence: i64,
|
||||
pub term: i64,
|
||||
pub payload: serde_json::Value,
|
||||
}
|
||||
|
||||
impl Envelope {
|
||||
pub fn validate(&self) -> Result<Plane, EnvelopeError> {
|
||||
if self.protocol_version != PROTOCOL_VERSION {
|
||||
return Err(EnvelopeError::UnsupportedProtocol(self.protocol_version));
|
||||
}
|
||||
if self.source_node_id.is_empty() {
|
||||
return Err(EnvelopeError::EmptySourceNode);
|
||||
}
|
||||
if self.sequence < 0 {
|
||||
return Err(EnvelopeError::NegativeSequence(self.sequence));
|
||||
}
|
||||
if self.term < 0 {
|
||||
return Err(EnvelopeError::NegativeTerm(self.term));
|
||||
}
|
||||
self.message_type.source_plane()
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<String> for ProtocolVersion {
|
||||
type Error = EnvelopeError;
|
||||
|
||||
fn try_from(value: String) -> Result<Self, Self::Error> {
|
||||
match value.as_str() {
|
||||
"1.0" => Ok(PROTOCOL_VERSION),
|
||||
_ => Err(EnvelopeError::UnsupportedProtocolWire(value)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<ProtocolVersion> for String {
|
||||
fn from(value: ProtocolVersion) -> Self {
|
||||
format!("{}.0", value.0)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum EnvelopeError {
|
||||
#[error("message type must not be empty")]
|
||||
EmptyMessageType,
|
||||
#[error("source node id must not be empty")]
|
||||
EmptySourceNode,
|
||||
#[error("sequence must not be negative: {0}")]
|
||||
NegativeSequence(i64),
|
||||
#[error("term must not be negative: {0}")]
|
||||
NegativeTerm(i64),
|
||||
#[error("unsupported protocol version: {}", .0.0)]
|
||||
UnsupportedProtocol(ProtocolVersion),
|
||||
#[error("unsupported protocol version: {0}")]
|
||||
UnsupportedProtocolWire(String),
|
||||
#[error("unknown source plane: {0}")]
|
||||
UnknownSourcePlane(String),
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn envelope() -> Envelope {
|
||||
Envelope {
|
||||
message_id: Uuid::nil(),
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
schema_version: SchemaVersion(1),
|
||||
source_node_id: "node-1".to_owned(),
|
||||
repository_id: Uuid::nil(),
|
||||
message_type: MessageType(String::from("repository.ref.updated")),
|
||||
sequence: 7,
|
||||
term: 3,
|
||||
payload: serde_json::Value::Null,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_a_repository_envelope() {
|
||||
assert!(matches!(envelope().validate(), Ok(Plane::Repository)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_a_future_protocol() {
|
||||
let mut subject = envelope();
|
||||
subject.protocol_version = ProtocolVersion(PROTOCOL_VERSION.0 + 1);
|
||||
assert!(subject.validate().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_an_unknown_source_plane() {
|
||||
let mut subject = envelope();
|
||||
subject.message_type = MessageType(String::from("forge.ref.updated"));
|
||||
assert!(subject.validate().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_the_repository_wire_envelope_without_extensions()
|
||||
-> Result<(), Box<dyn std::error::Error>> {
|
||||
let json = r#"{"message_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89211","protocol_version":"1.0","schema_version":1,"source_node_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89212","repository_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89214","message_type":"repository.ref.updated","sequence":2,"term":1,"payload":{}}"#;
|
||||
let envelope: Envelope = serde_json::from_str(json)?;
|
||||
|
||||
assert!(matches!(envelope.validate(), Ok(Plane::Repository)));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn identifies_check_events_as_control_events() {
|
||||
let mut subject = envelope();
|
||||
subject.message_type = MessageType(String::from("check.status.changed"));
|
||||
assert!(matches!(subject.validate(), Ok(Plane::Control)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_unknown_fields() {
|
||||
let json = r#"{"message_id":"00000000-0000-0000-0000-000000000000","extra":1}"#;
|
||||
assert!(serde_json::from_str::<Envelope>(json).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_an_empty_message_type() {
|
||||
assert!(MessageType::new("").is_err());
|
||||
}
|
||||
}
|
||||
@@ -1,91 +1,109 @@
|
||||
use async_trait::async_trait;
|
||||
use sqlx::postgres::PgPool;
|
||||
use syncode_collab_application::{EventInbox, StoreError};
|
||||
use syncode_collab_model::Envelope;
|
||||
use uuid::Uuid;
|
||||
|
||||
pub struct PostgresInbox {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresInbox {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
}
|
||||
|
||||
fn unavailable(error: sqlx::Error) -> StoreError {
|
||||
StoreError::Unavailable(error.to_string())
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventInbox for PostgresInbox {
|
||||
async fn already_applied(&self, message_id: Uuid) -> Result<bool, StoreError> {
|
||||
let found: Option<(Uuid,)> =
|
||||
sqlx::query_as("SELECT message_id FROM event_inbox WHERE message_id = $1")
|
||||
.bind(message_id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(unavailable)?;
|
||||
Ok(found.is_some())
|
||||
}
|
||||
|
||||
async fn last_position(
|
||||
&self,
|
||||
source_node_id: &str,
|
||||
resource_id: Uuid,
|
||||
) -> Result<Option<(i64, i64)>, StoreError> {
|
||||
sqlx::query_as(
|
||||
"SELECT last_sequence, last_term FROM event_inbox_position \
|
||||
WHERE source_node_id = $1 AND resource_id = $2",
|
||||
)
|
||||
.bind(source_node_id)
|
||||
.bind(resource_id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(unavailable)
|
||||
}
|
||||
|
||||
/// The deduplication key and the position advance share one transaction, so
|
||||
/// a crash between them cannot let the same fact apply twice.
|
||||
async fn apply(&self, envelope: &Envelope) -> Result<(), StoreError> {
|
||||
let mut transaction = self.pool.begin().await.map_err(unavailable)?;
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO event_inbox (message_id, source_node_id, source_plane, resource_id, \
|
||||
message_type, schema_version, sequence, term, payload) \
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
|
||||
)
|
||||
.bind(envelope.message_id)
|
||||
.bind(&envelope.source_node_id)
|
||||
.bind(&envelope.source_plane)
|
||||
.bind(envelope.resource_id)
|
||||
.bind(envelope.message_type.as_str())
|
||||
.bind(i32::try_from(envelope.schema_version.0).unwrap_or(i32::MAX))
|
||||
.bind(envelope.sequence)
|
||||
.bind(envelope.term)
|
||||
.bind(&envelope.payload)
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.map_err(|error| StoreError::Rejected(error.to_string()))?;
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO event_inbox_position \
|
||||
(source_node_id, resource_id, last_sequence, last_term) \
|
||||
VALUES ($1, $2, $3, $4) \
|
||||
ON CONFLICT (source_node_id, resource_id) DO UPDATE \
|
||||
SET last_sequence = EXCLUDED.last_sequence, \
|
||||
last_term = EXCLUDED.last_term, \
|
||||
updated_at = now()",
|
||||
)
|
||||
.bind(&envelope.source_node_id)
|
||||
.bind(envelope.resource_id)
|
||||
.bind(envelope.sequence)
|
||||
.bind(envelope.term)
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.map_err(|error| StoreError::Rejected(error.to_string()))?;
|
||||
|
||||
transaction.commit().await.map_err(unavailable)
|
||||
}
|
||||
}
|
||||
use async_trait::async_trait;
|
||||
use sqlx::postgres::PgPool;
|
||||
use syncode_collab_application::{EventInbox, StoreError};
|
||||
use syncode_collab_model::Envelope;
|
||||
use uuid::Uuid;
|
||||
|
||||
pub struct PostgresInbox {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresInbox {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
}
|
||||
|
||||
fn unavailable(error: sqlx::Error) -> StoreError {
|
||||
StoreError::Unavailable(error.to_string())
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventInbox for PostgresInbox {
|
||||
async fn already_applied(&self, message_id: Uuid) -> Result<bool, StoreError> {
|
||||
let found: Option<(Uuid,)> =
|
||||
sqlx::query_as("SELECT message_id FROM event_inbox WHERE message_id = $1")
|
||||
.bind(message_id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(unavailable)?;
|
||||
Ok(found.is_some())
|
||||
}
|
||||
|
||||
async fn last_position(
|
||||
&self,
|
||||
source_node_id: &str,
|
||||
resource_id: Uuid,
|
||||
) -> Result<Option<(i64, i64)>, StoreError> {
|
||||
sqlx::query_as(
|
||||
"SELECT last_sequence, last_term FROM event_inbox_position \
|
||||
WHERE source_node_id = $1 AND resource_id = $2",
|
||||
)
|
||||
.bind(source_node_id)
|
||||
.bind(resource_id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(unavailable)
|
||||
}
|
||||
|
||||
/// The deduplication key and the position advance share one transaction, so
|
||||
/// a crash between them cannot let the same fact apply twice.
|
||||
async fn apply(&self, envelope: &Envelope) -> Result<(), StoreError> {
|
||||
let source_plane = envelope
|
||||
.message_type
|
||||
.source_plane()
|
||||
.map_err(|error| StoreError::Rejected(error.to_string()))?;
|
||||
let mut transaction = self.pool.begin().await.map_err(unavailable)?;
|
||||
|
||||
let inserted = sqlx::query(
|
||||
"INSERT INTO event_inbox (message_id, source_node_id, source_plane, resource_id, \
|
||||
message_type, schema_version, sequence, term, payload) \
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) \
|
||||
ON CONFLICT (message_id) DO NOTHING",
|
||||
)
|
||||
.bind(envelope.message_id)
|
||||
.bind(&envelope.source_node_id)
|
||||
.bind(source_plane.as_str())
|
||||
.bind(envelope.repository_id)
|
||||
.bind(envelope.message_type.as_str())
|
||||
.bind(i32::try_from(envelope.schema_version.0).unwrap_or(i32::MAX))
|
||||
.bind(envelope.sequence)
|
||||
.bind(envelope.term)
|
||||
.bind(&envelope.payload)
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.map_err(|error| StoreError::Rejected(error.to_string()))?;
|
||||
if inserted.rows_affected() == 0 {
|
||||
transaction.commit().await.map_err(unavailable)?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let advanced = sqlx::query(
|
||||
"INSERT INTO event_inbox_position \
|
||||
(source_node_id, resource_id, last_sequence, last_term) \
|
||||
VALUES ($1, $2, $3, $4) \
|
||||
ON CONFLICT (source_node_id, resource_id) DO UPDATE \
|
||||
SET last_sequence = EXCLUDED.last_sequence, \
|
||||
last_term = EXCLUDED.last_term, \
|
||||
updated_at = now() \
|
||||
WHERE EXCLUDED.last_term > event_inbox_position.last_term \
|
||||
OR (EXCLUDED.last_term = event_inbox_position.last_term \
|
||||
AND EXCLUDED.last_sequence > event_inbox_position.last_sequence)",
|
||||
)
|
||||
.bind(&envelope.source_node_id)
|
||||
.bind(envelope.repository_id)
|
||||
.bind(envelope.sequence)
|
||||
.bind(envelope.term)
|
||||
.execute(&mut *transaction)
|
||||
.await
|
||||
.map_err(|error| StoreError::Rejected(error.to_string()))?;
|
||||
if advanced.rows_affected() == 0 {
|
||||
return Err(StoreError::Rejected(format!(
|
||||
"event position ({}, {}) is not newer than the stored position",
|
||||
envelope.term, envelope.sequence
|
||||
)));
|
||||
}
|
||||
|
||||
transaction.commit().await.map_err(unavailable)
|
||||
}
|
||||
}
|
||||
@@ -1,0 +1,46 @@
|
||||
use syncode_collab_application::{EventInbox, StoreError};
|
||||
use syncode_collab_model::envelope::{Envelope, MessageType, PROTOCOL_VERSION, SchemaVersion};
|
||||
use syncode_collab_store::{PostgresInbox, connect, migrate};
|
||||
use uuid::Uuid;
|
||||
|
||||
async fn store() -> Option<(PostgresInbox, sqlx::PgPool)> {
|
||||
let url = std::env::var("SYNCODE_COLLAB_TEST_DATABASE_URL").ok()?;
|
||||
let pool = connect(&url, 4).await.ok()?;
|
||||
migrate(&pool).await.ok()?;
|
||||
sqlx::query("TRUNCATE event_inbox, event_inbox_position")
|
||||
.execute(&pool)
|
||||
.await
|
||||
.ok()?;
|
||||
Some((PostgresInbox::new(pool.clone()), pool))
|
||||
}
|
||||
|
||||
fn envelope(sequence: i64) -> Result<Envelope, syncode_collab_model::EnvelopeError> {
|
||||
Ok(Envelope {
|
||||
message_id: Uuid::new_v4(),
|
||||
protocol_version: PROTOCOL_VERSION,
|
||||
schema_version: SchemaVersion(1),
|
||||
source_node_id: "source-1".to_owned(),
|
||||
repository_id: Uuid::nil(),
|
||||
message_type: MessageType::new("repository.ref.updated")?,
|
||||
sequence,
|
||||
term: 1,
|
||||
payload: serde_json::json!({}),
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn refuses_to_move_an_inbox_position_backwards() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let Some((store, pool)) = store().await else {
|
||||
return Ok(());
|
||||
};
|
||||
assert!(store.apply(&envelope(2)?).await.is_ok());
|
||||
assert!(matches!(
|
||||
store.apply(&envelope(1)?).await,
|
||||
Err(StoreError::Rejected(_))
|
||||
));
|
||||
let positions: i64 = sqlx::query_scalar("SELECT count(*) FROM event_inbox")
|
||||
.fetch_one(&pool)
|
||||
.await?;
|
||||
assert_eq!(1, positions);
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user