Cargo.toml

@@ -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

README.md

@@ -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.

src/config.rs

@@ -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,
+}

src/http.rs

@@ -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(())
+    }
+}

src/runtime.rs

@@ -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),
+}

crates/collab-application/src/inbox.rs

@@ -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 },
+}

crates/collab-model/src/envelope.rs

@@ -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());
+    }
+}

crates/collab-store/src/inbox.rs

@@ -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)
+    }
+}

crates/collab-store/tests/inbox_postgres.rs

@@ -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(())
+}