refactor: split connection node from common for later endpoint

Closes #99

Cargo.toml

@@ -1,70 +1,5 @@
-[package]
-name = "autopush"
-version = "1.53.1"
-authors = [
-  "Ben Bangert <[email protected]>",
-  "JR Conlin <[email protected]>",
-  "Alex Crichton <[email protected]>",
-  "Phil Jenvey <[email protected]>",
+[workspace]
+members = [
+    "autopush-common",
+    "autopush",
 ]
-edition = "2018"
-
-[lib]
-name = "autopush"
-
-[[bin]]
-name = "autopush_rs"
-path = "src/main.rs"
-
-[dependencies]
-base64 = "0.10.0"
-bytes = "0.4.11"
-cadence = "0.16.0"
-chan-signal = "0.3.2"
-chrono = "0.4.6"
-config = "0.9.2"
-docopt = "1.0.2"
-env_logger = { version = "0.6.0", default-features = false }
-error-chain = "0.12.0"
-fernet = "0.1.0"
-futures = "0.1.25"
-futures-backoff = "0.1.0"
-hex = "0.3.2"
-httparse = "1.3.3"
-# XXX: pin to hyper 0.11 for now: 0.12 has many changes..
-hyper = "0.11.27"
-lazy_static = "1.2.0"
-libc = "0.2.46"
-log = { version = "0.4.6", features = ["max_level_info", "release_max_level_info"] }
-matches = "0.1.8"
-mozsvc-common = "0.1.0"
-openssl = "0.10.16"
-rand = "0.6.3"
-regex = "1.1.0"
-reqwest = "0.9.5"
-rusoto_core = "0.36.0"
-rusoto_credential = "0.15.0"
-rusoto_dynamodb = "0.36.0"
-sentry = { version = "0.13.0", features = ["with_error_chain"] }
-serde = "1.0.84"
-serde_derive = "1.0.84"
-serde_dynamodb = { git = "https://github.com/mockersf/serde_dynamodb", rev = "240974d591466c4f9a0831162a28d690f2f6e51a" }
-serde_json = "1.0.34"
-slog = { version = "2.4.1", features = ["max_level_trace", "release_max_level_info"] }
-slog-async = "2.3.0"
-slog-term = "2.4.0"
-slog-mozlog-json = "0.1.0"
-slog-scope = "4.1.1"
-slog-stdlog = "3.0.2"
-# state_machine_future = { version = "0.1.6", features = ["debug_code_generation"] }
-state_machine_future = "0.2.0"
-time = "0.1.41"
-tokio-core = "0.1.17"
-tokio-io = "0.1.10"
-tokio-openssl = "0.3.0"
-tokio-service = "0.1.0"
-tokio-tungstenite = { version = "0.6.0", default-features = false }
-tungstenite = { version = "0.6.1", default-features = false }
-uuid = { version = "0.7.1", features = ["serde", "v4"] }
-# XXX: pin woothee until >= 0.8.1
-woothee = "0.7.3"

Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.31.1-stretch as builder
+FROM rust:1.32.0-stretch as builder
 
 ADD . /app
 WORKDIR /app
@@ -8,7 +8,7 @@ RUN \
     cargo --version && \
     rustc --version && \
     mkdir -m 755 bin && \
-    cargo install --path . --root /app
+    cargo install --path autopush --root /app
 
 
 FROM debian:stretch-slim

autopush-common/Cargo.toml

@@ -0,0 +1,46 @@
+[package]
+name = "autopush_common"
+version = "1.0.0"
+authors = [
+  "Ben Bangert <[email protected]>",
+  "JR Conlin <[email protected]>",
+  "Alex Crichton <[email protected]>",
+  "Phil Jenvey <[email protected]>",
+]
+edition = "2018"
+
+[lib]
+name = "autopush_common"
+
+[dependencies]
+cadence = "0.16.0"
+chrono = "0.4.6"
+config = "0.9.2"
+error-chain = "0.12.0"
+futures = "0.1.25"
+futures-backoff = "0.1.0"
+httparse = "1.3.3"
+# XXX: pin to hyper 0.11 for now: 0.12 has many changes..
+hyper = "0.11.27"
+lazy_static = "1.2.0"
+mozsvc-common = "0.1.0"
+rand = "0.6.3"
+regex = "1.1.0"
+reqwest = "0.9.5"
+rusoto_core = "0.36.0"
+rusoto_credential = "0.15.0"
+rusoto_dynamodb = "0.36.0"
+sentry = { version = "0.13.0", features = ["with_error_chain"] }
+serde = "1.0.84"
+serde_derive = "1.0.84"
+serde_dynamodb = "0.2.1"
+serde_json = "1.0.34"
+slog = { version = "2.4.1", features = ["max_level_trace", "release_max_level_info"] }
+slog-async = "2.3.0"
+slog-term = "2.4.0"
+slog-mozlog-json = "0.1.0"
+slog-scope = "4.1.1"
+slog-stdlog = "3.0.2"
+tokio-core = "0.1.17"
+tungstenite = { version = "0.6.1", default-features = false }
+uuid = { version = "0.7.1", features = ["serde", "v4"] }

src/db/commands.rs → autopush-common/src/db/commands.rs

@@ -20,7 +20,7 @@ use super::models::{DynamoDbNotification, DynamoDbUser};
 use super::util::generate_last_connect;
 use super::{HelloResponse, MAX_EXPIRY, USER_RECORD_VERSION};
 use crate::errors::*;
-use crate::protocol::Notification;
+use crate::notification::Notification;
 use crate::util::timing::sec_since_epoch;
 
 macro_rules! retryable_error {

src/db/mod.rs → autopush-common/src/db/mod.rs

@@ -21,8 +21,7 @@ mod models;
 mod util;
 
 use crate::errors::*;
-use crate::protocol::Notification;
-use crate::server::{Server, ServerOptions};
+use crate::notification::Notification;
 use crate::util::timing::sec_since_epoch;
 
 use self::commands::{
@@ -58,7 +57,6 @@ pub struct CheckStorageResponse {
 
 pub enum RegisterResponse {
     Success { endpoint: String },
-
     Error { error_msg: String, status: u32 },
 }
 
@@ -71,7 +69,11 @@ pub struct DynamoStorage {
 }
 
 impl DynamoStorage {
-    pub fn from_opts(opts: &ServerOptions, metrics: StatsdClient) -> Result<Self> {
+    pub fn from_opts(
+        message_table_name: &str,
+        router_table_name: &str,
+        metrics: StatsdClient,
+    ) -> Result<Self> {
         let ddb: Box<dyn DynamoDb> = if let Ok(endpoint) = env::var("AWS_LOCAL_DYNAMODB") {
             Box::new(DynamoDbClient::new_with(
                 HttpClient::new().chain_err(|| "TLS initialization error")?,
@@ -86,7 +88,7 @@ impl DynamoStorage {
         };
         let ddb = Rc::new(ddb);
 
-        let mut message_table_names = list_message_tables(&ddb, &opts._message_table_name)
+        let mut message_table_names = list_message_tables(&ddb, &message_table_name)
             .map_err(|_| "Failed to locate message tables")?;
         // Valid message months are the current and last 2 months
         message_table_names.sort_unstable_by(|a, b| b.cmp(a));
@@ -100,7 +102,7 @@ impl DynamoStorage {
         Ok(Self {
             ddb,
             metrics: Rc::new(metrics),
-            router_table_name: opts._router_table_name.clone(),
+            router_table_name: router_table_name.to_owned(),
             message_table_names,
             current_message_month,
         })
@@ -194,23 +196,15 @@ impl DynamoStorage {
 
     pub fn register(
         &self,
-        srv: &Rc<Server>,
         uaid: &Uuid,
         channel_id: &Uuid,
         message_month: &str,
+        endpoint: &str,
         key: Option<String>,
     ) -> MyFuture<RegisterResponse> {
         let ddb = self.ddb.clone();
-        let endpoint = match srv.make_endpoint(uaid, channel_id, key) {
-            Ok(result) => result,
-            Err(_) => {
-                return Box::new(future::ok(RegisterResponse::Error {
-                    error_msg: "Failed to generate endpoint".to_string(),
-                    status: 400,
-                }));
-            }
-        };
         let mut chids = HashSet::new();
+        let endpoint = endpoint.to_owned();
         chids.insert(channel_id.to_hyphenated().to_string());
         let response = commands::save_channels(ddb, uaid, chids, message_month)
             .and_then(move |_| future::ok(RegisterResponse::Success { endpoint }))

src/db/models.rs → autopush-common/src/db/models.rs

@@ -10,7 +10,7 @@ use uuid::Uuid;
 
 use crate::db::util::generate_last_connect;
 use crate::errors::*;
-use crate::protocol::Notification;
+use crate::notification::Notification;
 use crate::util::timing::{ms_since_epoch, sec_since_epoch};
 
 use super::{MAX_EXPIRY, USER_RECORD_VERSION};

autopush-common/src/lib.rs

@@ -0,0 +1,16 @@
+#![recursion_limit = "1024"]
+
+#[macro_use]
+extern crate error_chain;
+#[macro_use]
+extern crate slog;
+#[macro_use]
+extern crate slog_scope;
+
+#[macro_use]
+pub mod db;
+pub mod errors;
+pub mod logging;
+pub mod notification;
+#[macro_use]
+pub mod util;

autopush-common/src/notification.rs

@@ -0,0 +1,66 @@
+//! Notification protocol
+use std::collections::HashMap;
+
+use serde_derive::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::util::ms_since_epoch;
+
+#[derive(Serialize, Default, Deserialize, Clone, Debug)]
+pub struct Notification {
+    #[serde(rename = "channelID")]
+    pub channel_id: Uuid,
+    pub version: String,
+    #[serde(default = "default_ttl", skip_serializing)]
+    pub ttl: u64,
+    #[serde(skip_serializing)]
+    pub topic: Option<String>,
+    #[serde(skip_serializing)]
+    pub timestamp: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub data: Option<String>,
+    #[serde(skip_serializing)]
+    pub sortkey_timestamp: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub headers: Option<HashMap<String, String>>,
+}
+
+impl Notification {
+    /// Return an appropriate sort_key to use for the chidmessageid
+    ///
+    /// For new messages:
+    ///     02:{sortkey_timestamp}:{chid}
+    ///
+    /// For topic messages:
+    ///     01:{chid}:{topic}
+    ///
+    /// Old format for non-topic messages that is no longer returned:
+    ///     {chid}:{message_id}
+    pub fn sort_key(&self) -> String {
+        let chid = self.channel_id.to_hyphenated();
+        if let Some(ref topic) = self.topic {
+            format!("01:{}:{}", chid, topic)
+        } else if let Some(sortkey_timestamp) = self.sortkey_timestamp {
+            format!(
+                "02:{}:{}",
+                if sortkey_timestamp == 0 {
+                    ms_since_epoch()
+                } else {
+                    sortkey_timestamp
+                },
+                chid
+            )
+        } else {
+            // Legacy messages which we should never get anymore
+            format!("{}:{}", chid, self.version)
+        }
+    }
+
+    pub fn expired(&self, at_sec: u64) -> bool {
+        at_sec >= self.timestamp as u64 + self.ttl as u64
+    }
+}
+
+fn default_ttl() -> u64 {
+    0
+}

src/util/mod.rs → autopush-common/src/util/mod.rs

@@ -6,16 +6,11 @@ use tokio_core::reactor::{Handle, Timeout};
 
 use crate::errors::*;
 
-pub mod megaphone;
-mod rc;
 mod send_all;
 pub mod timing;
-mod user_agent;
 
-pub use self::rc::RcObject;
 pub use self::send_all::MySendAll;
 pub use self::timing::{ms_since_epoch, sec_since_epoch, us_since_epoch};
-pub use self::user_agent::parse_user_agent;
 
 /// Convenience future to time out the resolution of `f` provided within the
 /// duration provided.

autopush/Cargo.toml

@@ -0,0 +1,57 @@
+[package]
+name = "autopush"
+version = "1.53.1"
+authors = [
+  "Ben Bangert <[email protected]>",
+  "JR Conlin <[email protected]>",
+  "Alex Crichton <[email protected]>",
+  "Phil Jenvey <[email protected]>",
+]
+edition = "2018"
+
+[[bin]]
+name = "autopush_rs"
+path = "src/main.rs"
+
+[dependencies]
+autopush_common = { path = "../autopush-common" }
+base64 = "0.10.0"
+bytes = "0.4.11"
+cadence = "0.16.0"
+chan-signal = "0.3.2"
+config = "0.9.2"
+docopt = "1.0.2"
+error-chain = "0.12.0"
+fernet = "0.1.0"
+futures = "0.1.25"
+hex = "0.3.2"
+httparse = "1.3.3"
+# XXX: pin to hyper 0.11 for now: 0.12 has many changes..
+hyper = "0.11.27"
+lazy_static = "1.2.0"
+mozsvc-common = "0.1.0"
+openssl = "0.10.16"
+reqwest = "0.9.5"
+rusoto_dynamodb = "0.36.0"
+sentry = { version = "0.13.0", features = ["with_error_chain"] }
+serde = "1.0.84"
+serde_derive = "1.0.84"
+serde_dynamodb = "0.2.1"
+serde_json = "1.0.34"
+slog = { version = "2.4.1", features = ["max_level_trace", "release_max_level_info"] }
+slog-async = "2.3.0"
+slog-term = "2.4.0"
+slog-mozlog-json = "0.1.0"
+slog-scope = "4.1.1"
+# state_machine_future = { version = "0.1.6", features = ["debug_code_generation"] }
+state_machine_future = "0.2.0"
+time = "0.1.41"
+tokio-core = "0.1.17"
+tokio-io = "0.1.10"
+tokio-openssl = "0.3.0"
+tokio-service = "0.1.0"
+tokio-tungstenite = { version = "0.6.0", default-features = false }
+tungstenite = { version = "0.6.1", default-features = false }
+uuid = { version = "0.7.1", features = ["serde", "v4"] }
+# XXX: pin woothee until >= 0.8.1
+woothee = "0.7.3"