feat: more task system

apps/recorder/src/task/config.rs

@@ -0,0 +1,4 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TaskConfig {}

apps/recorder/src/task/core.rs

@@ -0,0 +1,81 @@
+use std::sync::Arc;
+
+use futures::{Stream, TryStreamExt, pin_mut};
+use serde::{Serialize, de::DeserializeOwned};
+
+use crate::{
+    app::AppContextTrait,
+    errors::RecorderResult,
+    models::subscriber_tasks::{self, SubscriberTaskErrorSnapshot},
+};
+
+pub const SUBSCRIBER_TASK_APALIS_NAME: &str = "subscriber_task";
+
+#[async_trait::async_trait]
+pub trait SubscriberAsyncTaskTrait: Serialize + DeserializeOwned + Sized {
+    type Result: Serialize + DeserializeOwned + Send;
+
+    async fn run_async(
+        self,
+        ctx: Arc<dyn AppContextTrait>,
+        id: i32,
+    ) -> RecorderResult<Self::Result>;
+
+    async fn run(self, ctx: Arc<dyn AppContextTrait>, id: i32) -> RecorderResult<()> {
+        match self.run_async(ctx.clone(), id).await {
+            Ok(result) => {
+                subscriber_tasks::Model::update_result(ctx, id, result).await?;
+            }
+            Err(e) => {
+                let error_snapshot = SubscriberTaskErrorSnapshot {
+                    message: e.to_string(),
+                };
+
+                subscriber_tasks::Model::update_error(ctx, id, error_snapshot).await?;
+
+                return Err(e);
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+pub trait SubscriberStreamTaskTrait: Serialize + DeserializeOwned + Sized {
+    type Yield: Serialize + DeserializeOwned + Send;
+
+    fn run_stream(
+        self,
+        ctx: Arc<dyn AppContextTrait>,
+    ) -> impl Stream<Item = RecorderResult<Self::Yield>> + Send;
+
+    async fn run(self, ctx: Arc<dyn AppContextTrait>, id: i32) -> RecorderResult<()> {
+        let stream = self.run_stream(ctx.clone());
+
+        pin_mut!(stream);
+
+        loop {
+            match stream.try_next().await {
+                Ok(Some(result)) => {
+                    subscriber_tasks::Model::append_yield(ctx.clone(), id, result).await?;
+                }
+                Ok(None) => {
+                    subscriber_tasks::Model::update_result(ctx, id, ()).await?;
+                    break;
+                }
+                Err(e) => {
+                    let error_snapshot = SubscriberTaskErrorSnapshot {
+                        message: e.to_string(),
+                    };
+
+                    subscriber_tasks::Model::update_error(ctx, id, error_snapshot).await?;
+
+                    return Err(e);
+                }
+            }
+        }
+
+        Ok(())
+    }
+}

apps/recorder/src/task/mikan/mod.rs

@@ -0,0 +1,3 @@
+mod scrape_season_subscription;
+
+pub use scrape_season_subscription::MikanScrapeSeasonSubscriptionTask;

apps/recorder/src/task/mikan/scrape_season_subscription.rs

@@ -0,0 +1,45 @@
+use std::sync::Arc;
+
+use futures::Stream;
+use sea_orm::FromJsonQueryResult;
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    app::AppContextTrait,
+    errors::RecorderResult,
+    extract::mikan::{
+        MikanBangumiMeta, MikanSeasonStr, build_mikan_season_flow_url,
+        scrape_mikan_bangumi_meta_stream_from_season_flow_url,
+    },
+    task::SubscriberStreamTaskTrait,
+};
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, FromJsonQueryResult)]
+pub struct MikanScrapeSeasonSubscriptionTask {
+    pub task_id: i32,
+    pub year: i32,
+    pub season_str: MikanSeasonStr,
+    pub credential_id: i32,
+    pub subscriber_id: i32,
+}
+
+#[async_trait::async_trait]
+impl SubscriberStreamTaskTrait for MikanScrapeSeasonSubscriptionTask {
+    type Yield = MikanBangumiMeta;
+
+    fn run_stream(
+        self,
+        ctx: Arc<dyn AppContextTrait>,
+    ) -> impl Stream<Item = RecorderResult<Self::Yield>> {
+        let mikan_base_url = ctx.mikan().base_url().clone();
+
+        let mikan_season_flow_url =
+            build_mikan_season_flow_url(mikan_base_url, self.year, self.season_str);
+
+        scrape_mikan_bangumi_meta_stream_from_season_flow_url(
+            ctx.clone(),
+            mikan_season_flow_url,
+            self.credential_id,
+        )
+    }
+}

apps/recorder/src/task/mod.rs

@@ -0,0 +1,13 @@
+mod config;
+mod core;
+pub mod mikan;
+mod registry;
+mod service;
+
+pub use core::{SUBSCRIBER_TASK_APALIS_NAME, SubscriberAsyncTaskTrait, SubscriberStreamTaskTrait};
+
+pub use config::TaskConfig;
+pub use registry::{
+    SubscriberTask, SubscriberTaskPayload, SubscriberTaskType, SubscriberTaskTypeEnum,
+};
+pub use service::TaskService;

apps/recorder/src/task/registry.rs

@@ -0,0 +1,33 @@
+use sea_orm::{DeriveActiveEnum, DeriveDisplay, prelude::*};
+use serde::{Deserialize, Serialize};
+
+use super::mikan::MikanScrapeSeasonSubscriptionTask;
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, DeriveDisplay, Serialize, Deserialize,
+)]
+#[sea_orm(
+    rs_type = "String",
+    db_type = "String(StringLen::None)",
+    enum_name = "subscriber_task_type"
+)]
+#[serde(rename_all = "snake_case")]
+pub enum SubscriberTaskType {
+    #[sea_orm(string_value = "mikan_scrape_season_subscription")]
+    MikanScrapeSeasonSubscription,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "task_type")]
+pub enum SubscriberTaskPayload {
+    #[serde(rename = "mikan_scrape_season_subscription")]
+    MikanScrapeSeasonSubscription(MikanScrapeSeasonSubscriptionTask),
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+pub struct SubscriberTask {
+    pub id: i32,
+    pub subscriber_id: i32,
+    #[serde(flatten)]
+    pub payload: SubscriberTaskPayload,
+}

apps/recorder/src/task/service.rs

@@ -0,0 +1,77 @@
+use std::{ops::Deref, sync::Arc};
+
+use apalis::prelude::*;
+use apalis_sql::postgres::PostgresStorage;
+use tokio::sync::RwLock;
+
+use crate::{
+    app::AppContextTrait,
+    errors::RecorderResult,
+    task::{
+        SUBSCRIBER_TASK_APALIS_NAME, SubscriberStreamTaskTrait, SubscriberTask,
+        SubscriberTaskPayload, TaskConfig,
+    },
+};
+
+pub struct TaskService {
+    pub config: TaskConfig,
+    ctx: Arc<dyn AppContextTrait>,
+    subscriber_task_storage: Arc<RwLock<PostgresStorage<SubscriberTask>>>,
+}
+
+impl TaskService {
+    pub async fn from_config_and_ctx(
+        config: TaskConfig,
+        ctx: Arc<dyn AppContextTrait>,
+    ) -> RecorderResult<Self> {
+        let pool = ctx.db().get_postgres_connection_pool().clone();
+        let subscriber_task_storage = Arc::new(RwLock::new(PostgresStorage::new(pool)));
+
+        Ok(Self {
+            config,
+            ctx,
+            subscriber_task_storage,
+        })
+    }
+
+    async fn run_subscriber_task(
+        job: SubscriberTask,
+        data: Data<Arc<dyn AppContextTrait>>,
+    ) -> RecorderResult<()> {
+        let ctx = data.deref().clone();
+
+        match job.payload {
+            SubscriberTaskPayload::MikanScrapeSeasonSubscription(task) => {
+                task.run(ctx, job.id).await
+            }
+        }
+    }
+
+    pub async fn add_subscriber_task(&self, job: SubscriberTask) -> RecorderResult<()> {
+        {
+            let mut storage = self.subscriber_task_storage.write().await;
+            storage.push(job).await?;
+        }
+
+        Ok(())
+    }
+
+    pub async fn setup(&self) -> RecorderResult<()> {
+        let monitor = Monitor::new();
+        let worker = WorkerBuilder::new(SUBSCRIBER_TASK_APALIS_NAME)
+            .catch_panic()
+            .enable_tracing()
+            .data(self.ctx.clone())
+            .backend({
+                let storage = self.subscriber_task_storage.read().await;
+                storage.clone()
+            })
+            .build_fn(Self::run_subscriber_task);
+
+        let monitor = monitor.register(worker);
+
+        monitor.run().await?;
+
+        Ok(())
+    }
+}