Simplify process_task_inner in worker (#934)

## Motivation

This is a functional NOOP refactor the worker, that largely aims to
- Remove uses of `unreachable!` macro
- Remove unnecessary messagedestination fetch attempt in dispatch
function
- Just generally simplify things, remove code, etc.

## Solution
- This mostly refactors `process_queue_task_inner` so that the task type
is `match`ed on only once. Overall this cuts down on code duplication
too.

Author: svix-gabriel

server/svix-server/src/worker.rs

@@ -19,9 +19,7 @@ use crate::core::webhook_http_client::{
 };
 use crate::db::models::{endpoint, message, messageattempt, messagedestination};
 use crate::error::{Error, ErrorType, HttpError, Result};
-use crate::queue::{
-    MessageTask, MessageTaskBatch, QueueTask, TaskQueueConsumer, TaskQueueProducer,
-};
+use crate::queue::{MessageTask, QueueTask, TaskQueueConsumer, TaskQueueProducer};
 use crate::v1::utils::get_unix_timestamp;
 use crate::{ctx, err_cache, err_generic, err_validation};
 
@@ -35,7 +33,8 @@ use rand::Rng;
 
 use sea_orm::prelude::DateTimeUtc;
 use sea_orm::{
-    ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set, TryIntoModel,
+    ActiveModelBehavior, ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait,
+    QueryFilter, Set, TryIntoModel,
 };
 use serde::{Deserialize, Serialize};
 use tokio::time::sleep;
@@ -644,24 +643,12 @@ async fn dispatch_message_task(
     msg_task: MessageTask,
     payload: &str,
     endp: CreateMessageEndpoint,
-    msg_dest: Option<messagedestination::Model>,
+    msg_dest: messagedestination::Model,
 ) -> Result<()> {
-    let WorkerContext { cfg, db, .. } = worker_context;
+    let WorkerContext { cfg, .. } = worker_context;
 
     tracing::trace!("Dispatch start");
 
-    let msg_dest = if let Some(msg_dest) = msg_dest {
-        msg_dest
-    } else {
-        ctx!(
-            messagedestination::Entity::secure_find_by_msg(msg_task.msg_id.clone())
-                .filter(messagedestination::Column::EndpId.eq(endp.id.clone()))
-                .one(*db)
-                .await
-        )?
-        .ok_or_else(|| err_generic!("Msg dest not found {} {}", msg_task.msg_id, endp.id))?
-    };
-
     if (msg_dest.status != MessageStatus::Pending && msg_dest.status != MessageStatus::Sending)
         && (msg_task.trigger_type != MessageAttemptTriggerType::Manual)
     {
@@ -735,32 +722,50 @@ async fn process_queue_task_inner(
     queue_task: QueueTask,
 ) -> Result<()> {
     let WorkerContext { db, cache, .. }: WorkerContext<'_> = worker_context;
+    let span = tracing::Span::current();
 
-    if queue_task == QueueTask::HealthCheck {
-        return Ok(());
-    }
+    let (msg, force_endpoint, destination, trigger_type, attempt_count) = match queue_task {
+        QueueTask::HealthCheck => return Ok(()),
+        QueueTask::MessageV1(task) => {
+            let msg = ctx!(
+                message::Entity::find_by_id(task.msg_id.clone())
+                    .one(db)
+                    .await
+            )?
+            .ok_or_else(|| err_generic!("Unexpected: message doesn't exist"))?;
 
-    let span = tracing::Span::current();
+            let destination = ctx!(
+                messagedestination::Entity::secure_find_by_msg(task.msg_id.clone())
+                    .filter(messagedestination::Column::EndpId.eq(task.endpoint_id.clone()))
+                    .one(db)
+                    .await
+            )?
+            .ok_or_else(|| {
+                err_generic!(format!(
+                    "MessageDestination not found for message {}",
+                    &task.msg_id
+                ))
+            })?;
 
-    let (msg_id, trigger_type) = match &queue_task {
-        QueueTask::MessageBatch(MessageTaskBatch {
-            msg_id,
-            trigger_type,
-            ..
-        }) => (msg_id, trigger_type),
-        QueueTask::MessageV1(MessageTask {
-            msg_id,
-            trigger_type,
-            ..
-        }) => (msg_id, trigger_type),
-
-        QueueTask::HealthCheck => unreachable!(),
+            (
+                msg,
+                Some(task.endpoint_id),
+                Some(destination),
+                task.trigger_type,
+                task.attempt_count,
+            )
+        }
+        QueueTask::MessageBatch(task) => {
+            let msg = ctx!(message::Entity::find_by_id(task.msg_id).one(db).await)?
+                .ok_or_else(|| err_generic!("Unexpected: message doesn't exist"))?;
+            (msg, None, None, task.trigger_type, 0)
+        }
     };
 
-    span.record("msg_id", &msg_id.0);
+    span.record("msg_id", &msg.id.0);
+    span.record("app_id", &msg.app_id.0);
+    span.record("org_id", &msg.org_id.0);
 
-    let msg = ctx!(message::Entity::find_by_id(msg_id.clone()).one(db).await)?
-        .ok_or_else(|| err_generic!("Unexpected: message doesn't exist"))?;
     let payload = match msg
         .payload
         .as_ref()
@@ -773,9 +778,6 @@ async fn process_queue_task_inner(
         }
     };
 
-    span.record("app_id", &msg.app_id.0);
-    span.record("org_id", &msg.org_id.0);
-
     let create_message_app = match CreateMessageApp::layered_fetch(
         cache,
         db,
@@ -794,60 +796,26 @@ async fn process_queue_task_inner(
     };
 
     let endpoints: Vec<CreateMessageEndpoint> = create_message_app
-        .filtered_endpoints(*trigger_type, &msg.event_type, msg.channels.as_ref())
+        .filtered_endpoints(trigger_type, &msg.event_type, msg.channels.as_ref())
         .iter()
-        .filter(|endpoint| match &queue_task {
-            QueueTask::HealthCheck => unreachable!(),
-            QueueTask::MessageV1(task) => task.endpoint_id == endpoint.id,
-            QueueTask::MessageBatch(_) => true,
+        .filter(|endpoint| match force_endpoint.as_ref() {
+            Some(endp_id) => endp_id == &endpoint.id,
+            None => true,
         })
         .cloned()
         .collect();
 
-    let futures: Vec<_> = match &queue_task {
-        QueueTask::HealthCheck => unreachable!(),
-
-        QueueTask::MessageV1(task) => {
-            let endpoint = match endpoints.into_iter().next() {
-                Some(ep) => ep,
-                None => {
-                    return Ok(());
-                }
-            };
-
-            let destination = ctx!(
-                messagedestination::Entity::secure_find_by_msg(task.msg_id.clone())
-                    .filter(messagedestination::Column::EndpId.eq(endpoint.id.clone()))
-                    .one(db)
-                    .await
-            )?
-            .ok_or_else(|| {
-                err_generic!(format!(
-                    "MessageDestination not found for message {}",
-                    &task.msg_id
-                ))
-            })?;
-
-            vec![dispatch_message_task(
-                &worker_context,
-                &msg,
-                &create_message_app,
-                task.clone(),
-                &payload,
-                endpoint,
-                Some(destination),
-            )]
-        }
-
-        QueueTask::MessageBatch(task) => {
+    let destinations = match destination {
+        Some(d) => vec![d],
+        None => {
             let destinations: Vec<_> = endpoints
                 .iter()
                 .map(|endpoint| messagedestination::ActiveModel {
                     msg_id: Set(msg.id.clone()),
                     endp_id: Set(endpoint.id.clone()),
                     next_attempt: Set(Some(Utc::now().into())),
                     status: Set(MessageStatus::Sending),
-                    ..Default::default()
+                    ..messagedestination::ActiveModel::new()
                 })
                 .collect();
 
@@ -857,32 +825,37 @@ async fn process_queue_task_inner(
                     .await
             )?;
 
-            endpoints
+            let dests: std::result::Result<_, _> = destinations
                 .into_iter()
-                .zip(destinations)
-                .map(|(endpoint, destination)| {
-                    let task = MessageTask {
-                        msg_id: msg_id.clone(),
-                        app_id: task.app_id.clone(),
-                        endpoint_id: endpoint.id.clone(),
-                        attempt_count: 0,
-                        trigger_type: *trigger_type,
-                    };
-
-                    dispatch_message_task(
-                        &worker_context,
-                        &msg,
-                        &create_message_app,
-                        task,
-                        &payload,
-                        endpoint,
-                        destination.try_into_model().ok(),
-                    )
-                })
-                .collect()
+                .map(|d| d.try_into_model())
+                .collect();
+            ctx!(dests)?
         }
     };
 
+    let futures = endpoints
+        .into_iter()
+        .zip(destinations)
+        .map(|(endpoint, destination)| {
+            let task = MessageTask {
+                msg_id: msg.id.clone(),
+                app_id: create_message_app.id.clone(),
+                endpoint_id: endpoint.id.clone(),
+                attempt_count,
+                trigger_type,
+            };
+
+            dispatch_message_task(
+                &worker_context,
+                &msg,
+                &create_message_app,
+                task,
+                &payload,
+                endpoint,
+                destination,
+            )
+        });
+
     let join = future::join_all(futures).await;
 
     let errs: Vec<_> = join.iter().filter(|x| x.is_err()).collect();