feat: validate schedule input

src/contract.rs

@@ -7,12 +7,12 @@ use chrono::prelude::*;
 use mockall::{automock, mock, predicate::*};
 use uuid::Uuid;
 
-use crate::model::{Message, MetricEvent, Schedule, Transmission};
+use crate::model::{Message, MetricEvent, Schedule, ScheduleError, Transmission};
 
 #[cfg_attr(test, automock)]
 #[async_trait]
 pub trait Scheduler: Send + Sync {
-    async fn schedule(&self, schedule: Schedule, message: Message) -> Result<Uuid, Box<dyn Error>>;
+    async fn schedule(&self, schedule: Schedule, message: Message) -> Result<Uuid, ScheduleError>;
 }
 
 #[cfg_attr(test, automock)]

src/grpc.rs

@@ -243,6 +243,15 @@ impl proto::transmit_server::Transmit for GrpcServer {
                     transmission_id: id.to_string(),
                 }))
             }
+            Err(ScheduleError::AgedSchedule) => Err(Status::invalid_argument(
+                "aged schedule; provide a timestamp in the future",
+            )),
+            Err(ScheduleError::TooShortInterval) => Err(Status::invalid_argument(
+                "too short interval; provide a greater duration between transmissions",
+            )),
+            Err(ScheduleError::NatsInvalidSubject) => Err(Status::invalid_argument(
+                "provided nats subject not allowed",
+            )),
             Err(err) => {
                 error!("Failed to schedule message: {err}");
 

src/integration_test.rs

@@ -53,6 +53,13 @@ mod tests {
 
         let mut now = MockNow::new();
         let mut sequence_now = Sequence::new();
+
+        // Expect call for validation.
+        now.expect_now()
+            .once()
+            .in_sequence(&mut sequence_now)
+            .returning(move || timestamp_now);
+
         for process_listen_iteration in process_listen_iterations.clone() {
             now.expect_now()
                 .once()
@@ -136,6 +143,13 @@ mod tests {
 
         let mut now = MockNow::new();
         let mut sequence_now = Sequence::new();
+
+        // Expect call for validation.
+        now.expect_now()
+            .once()
+            .in_sequence(&mut sequence_now)
+            .returning(move || timestamp_now);
+
         for process_listen_iteration in process_listen_iterations.clone() {
             now.expect_now()
                 .once()
@@ -234,6 +248,13 @@ mod tests {
 
         let mut now = MockNow::new();
         let mut sequence_now = Sequence::new();
+
+        // Expect call for validation.
+        now.expect_now()
+            .once()
+            .in_sequence(&mut sequence_now)
+            .returning(move || timestamp_now);
+
         for process_listen_iteration in process_listen_iterations.clone() {
             now.expect_now()
                 .once()

src/model.rs

@@ -1,4 +1,5 @@
 use std::error::Error;
+use std::fmt;
 use std::str::FromStr;
 use std::time;
 
@@ -98,7 +99,7 @@ impl Schedule {
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
 pub struct Delayed {
-    transmit_at: DateTime<Utc>,
+    pub transmit_at: DateTime<Utc>,
 }
 
 impl Delayed {
@@ -221,6 +222,49 @@ pub enum Iterate {
     Times(u32),
 }
 
+#[derive(Debug)]
+pub enum ScheduleError {
+    AgedSchedule,
+    TooShortInterval,
+    NatsInvalidSubject,
+    Other(Box<dyn Error>),
+}
+
+impl fmt::Display for ScheduleError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            ScheduleError::AgedSchedule => {
+                write!(f, "first transmission should not be in the past")
+            }
+            ScheduleError::TooShortInterval => write!(f, "interval must be sufficiently large"),
+            ScheduleError::NatsInvalidSubject => {
+                write!(f, "subject not allowed")
+            }
+            ScheduleError::Other(err) => write!(f, "err: {}", err),
+        }
+    }
+}
+
+impl PartialEq for ScheduleError {
+    fn eq(&self, other: &Self) -> bool {
+        match self {
+            ScheduleError::AgedSchedule => match other {
+                ScheduleError::AgedSchedule => true,
+                _ => false,
+            },
+            ScheduleError::TooShortInterval => match other {
+                ScheduleError::TooShortInterval => true,
+                _ => false,
+            },
+            ScheduleError::NatsInvalidSubject => match other {
+                ScheduleError::NatsInvalidSubject => true,
+                _ => false,
+            },
+            _ => false,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Hash, PartialEq, Eq)]
 pub enum MetricEvent {
     Scheduled(bool),