Start of a crawl_feed job

src/bin/crawler.rs

@@ -5,8 +5,7 @@ use apalis_cron::{CronStream, Schedule};
 use apalis_redis::RedisStorage;
 use chrono::{DateTime, Utc};
 use clap::Parser;
-use lib::jobs::AsyncJob;
-use lib::models::feed::{Feed, GetFeedsOptions};
+use dotenvy::dotenv;
 use sqlx::postgres::PgPoolOptions;
 use sqlx::PgPool;
 use std::str::FromStr;
@@ -14,9 +13,10 @@ use std::sync::Arc;
 use thiserror::Error;
 use tracing::{info, instrument};
 
-use dotenvy::dotenv;
 use lib::config::Config;
+use lib::jobs::{AsyncJob, CrawlFeedJob};
 use lib::log::init_worker_tracing;
+use lib::models::feed::{Feed, GetFeedsOptions};
 
 #[derive(Default, Debug, Clone)]
 struct Crawl(DateTime<Utc>);
@@ -64,8 +64,9 @@ pub async fn crawl_fn(job: Crawl, state: Data<Arc<State>>) -> Result<(), CrawlEr
 
         for feed in feeds.into_iter() {
             // self.spawn_crawler_loop(feed, respond_to.clone());
+            // TODO: implement uniqueness on jobs per feed for ~1 minute
             apalis
-                .push(AsyncJob::HelloWorld(feed.feed_id.to_string()))
+                .push(AsyncJob::CrawlFeed(CrawlFeedJob { feed }))
                 .await
                 .map_err(|err| CrawlError::QueueJobError(err.to_string()))?;
         }

src/bin/worker.rs

@@ -1,17 +1,15 @@
 use anyhow::Result;
+use apalis::layers::retry::RetryPolicy;
 use apalis::layers::tracing::TraceLayer;
 use apalis::prelude::*;
 use apalis_redis::RedisStorage;
 use clap::Parser;
-
 use dotenvy::dotenv;
-use lib::config::Config;
-use lib::jobs::AsyncJob;
-use lib::log::init_worker_tracing;
+use tower::retry::RetryLayer;
 
-pub async fn worker_fn(job: AsyncJob) {
-    tracing::info!(job = ?job, "Hello, world!");
-}
+use lib::config::Config;
+use lib::jobs::{handle_async_job, AsyncJob};
+use lib::log::init_worker_tracing;
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -28,9 +26,10 @@ async fn main() -> Result<()> {
     Monitor::<TokioExecutor>::new()
         .register_with_count(2, {
             WorkerBuilder::new("worker")
+                .layer(RetryLayer::new(RetryPolicy::default()))
                 .layer(TraceLayer::new())
                 .backend(apalis_storage)
-                .build_fn(worker_fn)
+                .build_fn(handle_async_job)
         })
         .run()
         .await