Server: Worker refactor

server/Cargo.toml

@@ -1,2 +1,5 @@
 [workspace]
 members = ["svix-server", "svix-server_derive"]
+
+[patch.crates-io]
+hyper = { git = "https://github.com/svix/hyper/", rev = "b901ca7c" }

server/run-tests.sh

@@ -8,6 +8,7 @@ TEST_COMMAND="cargo test --all --all-features --all-targets"
 export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/postgres"
 export SVIX_JWT_SECRET="test value"
 export SVIX_LOG_LEVEL="info"
+export SVIX_WHITELIST_SUBNETS="[127.0.0.1/32]"
 
 echo "*********** RUN 1 ***********"
 SVIX_QUEUE_TYPE="redis" \

server/svix-server/Cargo.toml

@@ -23,6 +23,8 @@ clap = { version = "3.2.1", features = ["derive"] }
 axum = { version = "0.6.1", features = ["headers"] }
 base64 = "0.13.0"
 hyper = { version = "0.14.16", features = ["full"] }
+hyper-openssl = "0.9.2"
+openssl = "0.10.45"
 tokio = { version = "1.23.1", features = ["full"] }
 tower = "0.4.11"
 tower-http = { version = "0.3.4", features = ["trace", "cors", "request-id"] }
@@ -66,6 +68,11 @@ jsonschema = "0.16.1"
 aide = { version = "0.9.0", features = ["axum", "redoc", "macros"] }
 schemars = { version = "0.8.11", features = ["chrono", "url"] }
 indexmap = "1.9.2"
+trust-dns-resolver = "0.22.0"
+ipnet = { version = "2.5", features = ["serde"] }
+urlencoding = "2.1.2"
+strum_macros = "0.24"
+strum = { version = "0.24", features = ["derive"] }
 
 [dev-dependencies]
 anyhow = "1.0.56"

server/svix-server/config.default.toml

@@ -94,3 +94,11 @@ api_enabled = true
 
 # Should this instance run the message worker
 worker_enabled = true
+
+# Subnets to whitelist for outbound webhooks. Note that allowing endpoints in private IP space
+# is a security risk and should only be allowed if you are using the service internally or for
+# testing purposes. Should be specified in CIDR notation, e.g., `[127.0.0.1/32, 172.17.0.0/16, 192.168.0.0/16]`
+# whitelist_subnets = []
+
+# Maximum number of concurrent worker tasks to spawn (0 is unlimited)
+worker_max_tasks = 500

server/svix-server/src/cfg.rs

@@ -7,6 +7,7 @@ use figment::{
     providers::{Env, Format, Toml},
     Figment,
 };
+use ipnet::IpNet;
 use std::time::Duration;
 
 use crate::{core::cryptography::Encryption, core::security::Keys, error::Result};
@@ -173,6 +174,14 @@ pub struct ConfigurationInner {
     /// Should this instance run the message worker
     pub worker_enabled: bool,
 
+    /// Subnets to whitelist for outbound webhooks. Note that allowing endpoints in private IP space
+    /// is a security risk and should only be allowed if you are using the service internally or for
+    /// testing purposes. Should be specified in CIDR notation, e.g., `[127.0.0.1/32, 172.17.0.0/16, 192.168.0.0/16]`
+    pub whitelist_subnets: Option<Arc<Vec<IpNet>>>,
+
+    /// Maximum number of concurrent worker tasks to spawn (0 is unlimited)
+    pub worker_max_tasks: u16,
+
     #[serde(flatten)]
     pub internal: InternalConfig,
 }

server/svix-server/src/core/message_app.rs

@@ -67,7 +67,7 @@ impl CreateMessageApp {
     /// exists or from PostgreSQL otherwise. If the RedisCache is Some, but does not contain the
     /// requisite information, fetch it from PostgreSQL and insert the data into the cache.
     pub async fn layered_fetch(
-        cache: Cache,
+        cache: &Cache,
         pg: &DatabaseConnection,
         app: Option<application::Model>,
         org_id: OrganizationId,

server/svix-server/src/core/mod.rs

@@ -11,6 +11,7 @@ pub mod permissions;
 pub mod run_with_retries;
 pub mod security;
 pub mod types;
+pub mod webhook_http_client;
 
 #[cfg(test)]
 mod tests {