[Events] Add project notification settings

ghstack-source-id: 059cca5b0dd6e6645c83e4da43a0ec56fa738ffe
Pull Request resolved: #18


This PR adds a new model for project notifications. The semantics of the model is explained in `notifications.proto`.

1. Adds a the model to the db model in metadata_svc and exposed it in the project store.
2. Expose setters/getters to notification settings in metadata svc.
3. Expose APIs to manipulate the notification settings. I opted into having a single endpoint for updating the notification setting json for convenience.

Later, we can also add this to the CLI.

Test plan:

```
~/repos/cronback/cronback ❯❯❯ cargo cli --localhost --secret-token adminkey admin projects create                                                                                                              ✘ 130
Project 'prj_026601H699SE7VY0YEFHRXXE75117B' was created successfully.
~/repos/cronback/cronback ❯❯❯ http -b --auth adminkey --auth-type bearer POST http:https://localhost:8888/v1/admin/projects/prj_026601H699SE7VY0YEFHRXXE75117B/notification_settings @examples/notifications.json


~/repos/cronback/cronback ❯❯❯ http -b --auth adminkey --auth-type bearer GET http:https://localhost:8888/v1/admin/projects/prj_026601H699SE7VY0YEFHRXXE75117B/notification_settings
{
    "channels": {
        "email": {
            "address": "[email protected]",
            "type": "email",
            "verified": false
        }
    },
    "subscriptions": [
        {
            "channel_names": [
                "email"
            ],
            "event": {
                "type": "on_run_failure"
            }
        }
    ]
}
```

cronback-api-model/src/admin/mod.rs

@@ -1,5 +1,7 @@
 mod api_keys;
+mod notifications;
 mod projects;
 
 pub use api_keys::*;
+pub use notifications::*;
 pub use projects::*;

cronback-api-model/src/admin/notifications.rs

@@ -0,0 +1,240 @@
+use std::collections::HashMap;
+
+#[cfg(feature = "dto")]
+use dto::{FromProto, IntoProto};
+use monostate::MustBe;
+use serde::{Deserialize, Serialize};
+#[cfg(feature = "validation")]
+use validator::Validate;
+
+#[cfg(feature = "validation")]
+use crate::validation_util::validation_error;
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[cfg_attr(
+ feature = "dto",
+ derive(IntoProto, FromProto),
+ proto(target = "proto::notifications::ProjectNotificationSettings")
+)]
+#[cfg_attr(
+ feature = "validation",
+ derive(Validate),
+ validate(schema(
+ function = "validate_settings",
+ skip_on_field_errors = false
+ ))
+)]
+#[serde(deny_unknown_fields)]
+pub struct NotificationSettings {
+ #[cfg_attr(feature = "validation", validate)]
+ pub default_subscriptions: Vec<NotificationSubscription>,
+ // The key of the hashmap is the channel name
+ #[cfg_attr(feature = "validation", validate)]
+ pub channels: HashMap<String, NotificationChannel>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[cfg_attr(
+ feature = "dto",
+ derive(IntoProto, FromProto),
+ proto(target = "proto::notifications::NotificationSubscription")
+)]
+#[cfg_attr(feature = "validation", derive(Validate))]
+#[serde(deny_unknown_fields)]
+pub struct NotificationSubscription {
+ #[cfg_attr(feature = "validation", validate(length(max = 20)))]
+ pub channel_names: Vec<String>,
+ #[cfg_attr(feature = "dto", proto(required))]
+ #[cfg_attr(feature = "validation", validate)]
+ pub event: NotificationEvent,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[cfg_attr(
+ feature = "dto",
+ derive(IntoProto, FromProto),
+ proto(
+ target = "proto::notifications::NotificationChannel",
+ oneof = "channel"
+ )
+)]
+#[serde(rename_all = "snake_case")]
+#[serde(untagged)]
+pub enum NotificationChannel {
+ Email(EmailNotification),
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[cfg_attr(
+ feature = "dto",
+ derive(IntoProto, FromProto),
+ proto(target = "proto::notifications::NotificationEvent", oneof = "event")
+)]
+#[serde(rename_all = "snake_case")]
+#[serde(untagged)]
+pub enum NotificationEvent {
+ OnRunFailure(OnRunFailure),
+}
+
+// Channel configs
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[cfg_attr(
+ feature = "dto",
+ derive(IntoProto, FromProto),
+ proto(target = "proto::notifications::Email")
+)]
+#[serde(deny_unknown_fields)]
+#[cfg_attr(feature = "validation", derive(Validate))]
+pub struct EmailNotification {
+ #[serde(rename = "type")]
+ _kind: MustBe!("email"),
+ #[cfg_attr(feature = "validation", validate(email))]
+ pub address: String,
+ pub verified: bool,
+}
+
+// Subscription configs
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[cfg_attr(
+ feature = "dto",
+ derive(IntoProto, FromProto),
+ proto(target = "proto::notifications::OnRunFailure")
+)]
+#[cfg_attr(feature = "validation", derive(Validate))]
+#[serde(deny_unknown_fields)]
+pub struct OnRunFailure {
+ #[serde(rename = "type")]
+ _kind: MustBe!("on_run_failure"),
+}
+
+#[cfg(feature = "validation")]
+impl Validate for NotificationEvent {
+ fn validate(&self) -> Result<(), validator::ValidationErrors> {
+ match self {
+ | NotificationEvent::OnRunFailure(o) => o.validate(),
+ }
+ }
+}
+
+#[cfg(feature = "validation")]
+impl Validate for NotificationChannel {
+ fn validate(&self) -> Result<(), validator::ValidationErrors> {
+ match self {
+ | NotificationChannel::Email(e) => e.validate(),
+ }
+ }
+}
+
+#[cfg(feature = "validation")]
+fn validate_settings(
+ settings: &NotificationSettings,
+) -> Result<(), validator::ValidationError> {
+ // Validate that any channel referenced in a subscription actually exists.
+
+ for sub in &settings.default_subscriptions {
+ for channel in &sub.channel_names {
+ if !settings.channels.contains_key(channel) {
+ return Err(validation_error(
+ "invalid_channel_name",
+ format!(
+ "Channel name '{}' is not configured in channel list",
+ channel
+ ),
+ ));
+ }
+ }
+ }
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+
+ use super::*;
+
+ #[test]
+ fn test_valid_settings() -> anyhow::Result<()> {
+ let email = EmailNotification {
+ _kind: Default::default(),
+ address: "[email protected]".to_string(),
+ verified: true,
+ };
+ let mut channels = HashMap::new();
+ channels.insert("email".to_string(), NotificationChannel::Email(email));
+ let setting = NotificationSettings {
+ channels,
+ default_subscriptions: vec![NotificationSubscription {
+ channel_names: vec!["email".to_string()],
+ event: NotificationEvent::OnRunFailure(OnRunFailure {
+ _kind: Default::default(),
+ }),
+ }],
+ };
+
+ setting.validate()?;
+ Ok(())
+ }
+
+ #[test]
+ fn test_invalid_email() -> anyhow::Result<()> {
+ let email = EmailNotification {
+ _kind: Default::default(),
+ address: "wrong_email".to_string(),
+ verified: false,
+ };
+ let mut channels = HashMap::new();
+ channels.insert("email".to_string(), NotificationChannel::Email(email));
+ let setting = NotificationSettings {
+ channels,
+ default_subscriptions: vec![],
+ };
+
+ let validated = setting.validate();
+
+ assert!(validated.is_err());
+ assert_eq!(
+ validated.unwrap_err().to_string(),
+ "channels[0].address: Validation error: email [{\"value\": \
+ String(\"wrong_email\")}]"
+ .to_string()
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_invalid_channel() {
+ let email = EmailNotification {
+ _kind: Default::default(),
+ address: "[email protected]".to_string(),
+ verified: false,
+ };
+ let mut channels = HashMap::new();
+ channels.insert("email".to_string(), NotificationChannel::Email(email));
+ let setting = NotificationSettings {
+ channels,
+ default_subscriptions: vec![NotificationSubscription {
+ channel_names: vec![
+ "email".to_string(),
+ "wrong_channel".to_string(),
+ ],
+ event: NotificationEvent::OnRunFailure(OnRunFailure {
+ _kind: Default::default(),
+ }),
+ }],
+ };
+
+ let validated = setting.validate();
+
+ assert!(validated.is_err());
+ assert_eq!(
+ validated.unwrap_err().to_string(),
+ "__all__: Channel name 'wrong_channel' is not configured in \
+ channel list"
+ .to_string()
+ );
+ }
+}

cronback-dto-core/src/struct_codegen.rs

@@ -12,6 +12,7 @@ use crate::attributes::{
 };
 use crate::utils::{
  extract_inner_type_from_container,
+ map_segment,
  option_segment,
  vec_segment,
 };
@@ -59,6 +60,12 @@ impl ProtoFieldInfo {
  // - IntoProto + required: .into() should handle it.
  // - FromProto + required: our_name: incoming.unwrap()
  //
+ // - HashMap<K, V>:
+ // - Protobuf's map keys only support scaler types, we only need to map
+ // the values
+ // - IntoProto + required: .into() should handle it.
+ // - FromProto + required: our_name: incoming.unwrap()
+ //
  // - always add .into() after mapping.
 
  // Primary cases we need to take care of:
@@ -84,6 +91,7 @@ impl ProtoFieldInfo {
  let option_type =
  extract_inner_type_from_container(&self.ty, option_segment);
  let vec_type = extract_inner_type_from_container(&self.ty, vec_segment);
+ let map_type = extract_inner_type_from_container(&self.ty, map_segment);
 
  // 1. Do we need to unwrap the input before processing? We do that if
  // the field is `required` and our local type is not `Option<T>` when
@@ -163,6 +171,25 @@ impl ProtoFieldInfo {
  rhs_value_tok = quote_spanned! { span =>
  #rhs_value_tok.into_iter().map(#mapper).collect::<::std::vec::Vec<_>>()
  };
+ } else if let Some(_inner_ty) = map_type {
+ // A HashMap<K,V>
+ let mapper = self
+ .wrap_with_mapper(direction, quote! { v })
+ .map(|mapper| {
+ quote_spanned! { span =>
+ |(k, v)| (k, #mapper)
+ }
+ })
+ // If there is no mapper, we just map the inner value with any
+ // existing Into impl.
+ .unwrap_or_else(|| {
+ quote_spanned! { span =>
+ |(k, v)| (k, v.into())
+ }
+ });
+ rhs_value_tok = quote_spanned! { span =>
+ #rhs_value_tok.into_iter().map(#mapper).collect::<::std::collections::HashMap<_, _>>()
+ };
  } else {
  // Bare type
  rhs_value_tok = self

cronback-dto-core/src/utils.rs

@@ -29,6 +29,11 @@ pub(crate) fn vec_segment(path: &syn::Path) -> Option<&syn::PathSegment> {
  extract_generic_type_segment(path, VECTOR)
 }
 
+pub(crate) fn map_segment(path: &syn::Path) -> Option<&syn::PathSegment> {
+ static MAP: &[&str] = &["HashMap|", "std|collections|HashMap|"];
+ extract_generic_type_segment(path, MAP)
+}
+
 fn extract_type_path(ty: &syn::Type) -> Option<&syn::Path> {
  match *ty {
  | syn::Type::Path(ref typepath) if typepath.qself.is_none() => {

cronback-proto/build.rs

@@ -22,6 +22,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
  "./runs.proto",
  "./scheduler_svc.proto",
  "./triggers.proto",
+ "./notifications.proto",
  ],
  &["../proto"],
  )?;
@@ -36,6 +37,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
  ".events",
  ".projects",
  ".triggers",
+ ".notifications",
  ])?;
 
  Ok(())

cronback-proto/lib.rs

@@ -80,5 +80,9 @@ pub mod projects {
  include!(concat!(env!("OUT_DIR"), "/projects.serde.rs"));
 }
 
+pub mod notifications {
+ tonic::include_proto!("notifications");
+}
+
 pub const FILE_DESCRIPTOR_SET: &[u8] =
  tonic::include_file_descriptor_set!("file_descriptor");

cronback-proto/metadata_svc.proto

@@ -2,6 +2,7 @@ syntax = "proto3";
 
 import "common.proto";
 import "projects.proto";
+import "notifications.proto";
 
 package metadata_svc;
 
@@ -10,6 +11,8 @@ service MetadataSvc {
  rpc GetProjectStatus(GetProjectStatusRequest) returns (GetProjectStatusResponse);
  rpc SetProjectStatus(SetProjectStatusRequest) returns (SetProjectStatusResponse);
  rpc ProjectExists(ProjectExistsRequest) returns (ProjectExistsResponse);
+ rpc GetNotificationSettings(GetNotificationSettingsRequest) returns (GetNotificationSettingsResponse);
+ rpc SetNotificationSettings(SetNotificationSettingsRequest) returns (SetNotificationSettingsResponse);
 }
 
 message CreateProjectRequest {
@@ -45,3 +48,21 @@ message ProjectExistsResponse {
  bool exists = 1;
 }
 
+message GetNotificationSettingsRequest {
+ common.ProjectId id = 1;
+}
+
+message GetNotificationSettingsResponse {
+ notifications.ProjectNotificationSettings settings = 1;
+}
+
+
+message SetNotificationSettingsRequest {
+ common.ProjectId id = 1;
+ notifications.ProjectNotificationSettings settings = 2;
+}
+
+message SetNotificationSettingsResponse {
+ notifications.ProjectNotificationSettings old_settings = 1;
+}
+