feat: add basic webui

apps/recorder/src/models/bangumi.rs

@@ -0,0 +1,45 @@
+use loco_rs::app::AppContext;
+use sea_orm::{entity::prelude::*, ActiveValue, TryIntoModel};
+
+pub use super::entities::bangumi::*;
+
+impl Model {
+    pub async fn get_or_insert_from_mikan<F>(
+        ctx: &AppContext,
+        subscriber_id: i32,
+        subscription_id: i32,
+        mikan_bangumi_id: String,
+        mikan_fansub_id: String,
+        f: F,
+    ) -> eyre::Result<Model>
+    where
+        F: AsyncFnOnce(&mut ActiveModel) -> eyre::Result<()>,
+    {
+        let db = &ctx.db;
+        if let Some(existed) = Entity::find()
+            .filter(
+                Column::MikanBangumiId
+                    .eq(Some(mikan_bangumi_id.clone()))
+                    .and(Column::MikanFansubId.eq(Some(mikan_fansub_id.clone()))),
+            )
+            .one(db)
+            .await?
+        {
+            Ok(existed)
+        } else {
+            let mut bgm = ActiveModel {
+                mikan_bangumi_id: ActiveValue::Set(Some(mikan_bangumi_id)),
+                mikan_fansub_id: ActiveValue::Set(Some(mikan_fansub_id)),
+                subscription_id: ActiveValue::Set(subscription_id),
+                subscriber_id: ActiveValue::Set(subscriber_id),
+                ..Default::default()
+            };
+            f(&mut bgm).await?;
+            let bgm: Model = bgm.save(db).await?.try_into_model()?;
+            Ok(bgm)
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for ActiveModel {}

apps/recorder/src/models/downloaders.rs

@@ -0,0 +1,17 @@
+use sea_orm::prelude::*;
+use url::Url;
+
+pub use crate::models::entities::downloaders::*;
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for ActiveModel {}
+
+impl Model {
+    pub fn get_endpoint(&self) -> String {
+        self.endpoint.clone()
+    }
+    pub fn endpoint_url(&self) -> Result<Url, url::ParseError> {
+        let url = Url::parse(&self.endpoint)?;
+        Ok(url)
+    }
+}

apps/recorder/src/models/downloads.rs

@@ -0,0 +1,27 @@
+use sea_orm::{prelude::*, ActiveValue};
+
+use crate::extract::mikan::MikanRssItem;
+pub use crate::models::entities::downloads::*;
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for ActiveModel {}
+
+impl ActiveModel {
+    pub fn from_mikan_rss_item(m: MikanRssItem, subscription_id: i32) -> Self {
+        let _ = Self {
+            origin_name: ActiveValue::Set(m.title.clone()),
+            display_name: ActiveValue::Set(m.title),
+            subscription_id: ActiveValue::Set(subscription_id),
+            status: ActiveValue::Set(DownloadStatus::Pending),
+            mime: ActiveValue::Set(DownloadMime::BitTorrent),
+            url: ActiveValue::Set(m.url.to_string()),
+            curr_size: ActiveValue::Set(m.content_length.as_ref().map(|_| 0)),
+            all_size: ActiveValue::Set(m.content_length),
+            homepage: ActiveValue::Set(Some(m.homepage.to_string())),
+            ..Default::default()
+        };
+        todo!()
+    }
+}
+
+impl Model {}

apps/recorder/src/models/entities/bangumi.rs

@@ -0,0 +1,80 @@
+use sea_orm::{entity::prelude::*, FromJsonQueryResult};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, FromJsonQueryResult)]
+pub struct BangumiFilter {
+    pub name: Option<Vec<String>>,
+    pub group: Option<Vec<String>>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, FromJsonQueryResult)]
+pub struct BangumiExtra {
+    pub name_zh: Option<String>,
+    pub s_name_zh: Option<String>,
+    pub name_en: Option<String>,
+    pub s_name_en: Option<String>,
+    pub name_jp: Option<String>,
+    pub s_name_jp: Option<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "bangumi")]
+pub struct Model {
+    pub created_at: DateTime,
+    pub updated_at: DateTime,
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub mikan_bangumi_id: Option<String>,
+    pub subscription_id: i32,
+    pub subscriber_id: i32,
+    pub display_name: String,
+    pub raw_name: String,
+    pub season: i32,
+    pub season_raw: Option<String>,
+    pub fansub: Option<String>,
+    pub mikan_fansub_id: Option<String>,
+    pub filter: Option<BangumiFilter>,
+    pub rss_link: Option<String>,
+    pub poster_link: Option<String>,
+    pub save_path: Option<String>,
+    #[sea_orm(default = "false")]
+    pub deleted: bool,
+    pub homepage: Option<String>,
+    pub extra: Option<BangumiExtra>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::subscriptions::Entity",
+        from = "Column::SubscriptionId",
+        to = "super::subscriptions::Column::Id"
+    )]
+    Subscription,
+    #[sea_orm(
+        belongs_to = "super::subscribers::Entity",
+        from = "Column::SubscriberId",
+        to = "super::subscribers::Column::Id"
+    )]
+    Subscriber,
+    #[sea_orm(has_many = "super::episodes::Entity")]
+    Episode,
+}
+
+impl Related<super::episodes::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Episode.def()
+    }
+}
+
+impl Related<super::subscriptions::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscription.def()
+    }
+}
+
+impl Related<super::subscribers::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscriber.def()
+    }
+}

apps/recorder/src/models/entities/downloaders.rs

@@ -0,0 +1,45 @@
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, DeriveDisplay, Serialize, Deserialize,
+)]
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "downloader_type")]
+#[serde(rename_all = "snake_case")]
+pub enum DownloaderCategory {
+    #[sea_orm(string_value = "qbittorrent")]
+    QBittorrent,
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "downloaders")]
+pub struct Model {
+    #[sea_orm(column_type = "Timestamp")]
+    pub created_at: DateTime,
+    #[sea_orm(column_type = "Timestamp")]
+    pub updated_at: DateTime,
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub category: DownloaderCategory,
+    pub endpoint: String,
+    pub password: String,
+    pub username: String,
+    pub subscriber_id: i32,
+    pub save_path: String,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::subscribers::Entity",
+        from = "Column::SubscriberId",
+        to = "super::subscribers::Column::Id"
+    )]
+    Subscriber,
+}
+
+impl Related<super::subscribers::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscriber.def()
+    }
+}

apps/recorder/src/models/entities/downloads.rs

@@ -0,0 +1,78 @@
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, DeriveDisplay, Serialize, Deserialize,
+)]
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "download_status")]
+#[serde(rename_all = "snake_case")]
+pub enum DownloadStatus {
+    #[sea_orm(string_value = "pending")]
+    Pending,
+    #[sea_orm(string_value = "downloading")]
+    Downloading,
+    #[sea_orm(string_value = "paused")]
+    Paused,
+    #[sea_orm(string_value = "completed")]
+    Completed,
+    #[sea_orm(string_value = "failed")]
+    Failed,
+    #[sea_orm(string_value = "deleted")]
+    Deleted,
+}
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, DeriveDisplay, Serialize, Deserialize,
+)]
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "download_mime")]
+pub enum DownloadMime {
+    #[sea_orm(string_value = "application/octet-stream")]
+    #[serde(rename = "application/octet-stream")]
+    OctetStream,
+    #[sea_orm(string_value = "application/x-bittorrent")]
+    #[serde(rename = "application/x-bittorrent")]
+    BitTorrent,
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "downloads")]
+pub struct Model {
+    pub created_at: DateTime,
+    pub updated_at: DateTime,
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub origin_name: String,
+    pub display_name: String,
+    pub subscription_id: i32,
+    pub status: DownloadStatus,
+    pub mime: DownloadMime,
+    pub url: String,
+    pub all_size: Option<u64>,
+    pub curr_size: Option<u64>,
+    pub homepage: Option<String>,
+    pub save_path: Option<String>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::subscriptions::Entity",
+        from = "Column::SubscriptionId",
+        to = "super::subscriptions::Column::Id"
+    )]
+    Subscription,
+    #[sea_orm(has_many = "super::episodes::Entity")]
+    Episode,
+}
+
+impl Related<super::subscriptions::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscription.def()
+    }
+}
+
+impl Related<super::episodes::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Episode.def()
+    }
+}

apps/recorder/src/models/entities/episodes.rs

@@ -0,0 +1,95 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+use sea_orm::{entity::prelude::*, FromJsonQueryResult};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, FromJsonQueryResult, Default)]
+pub struct EpisodeExtra {
+    pub name_zh: Option<String>,
+    pub s_name_zh: Option<String>,
+    pub name_en: Option<String>,
+    pub s_name_en: Option<String>,
+    pub name_jp: Option<String>,
+    pub s_name_jp: Option<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "episodes")]
+pub struct Model {
+    pub created_at: DateTime,
+    pub updated_at: DateTime,
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    #[sea_orm(indexed)]
+    pub mikan_episode_id: Option<String>,
+    pub raw_name: String,
+    pub display_name: String,
+    pub bangumi_id: i32,
+    pub subscription_id: i32,
+    pub subscriber_id: i32,
+    pub download_id: Option<i32>,
+    pub save_path: Option<String>,
+    pub resolution: Option<String>,
+    pub season: i32,
+    pub season_raw: Option<String>,
+    pub fansub: Option<String>,
+    pub poster_link: Option<String>,
+    pub episode_index: i32,
+    pub homepage: Option<String>,
+    pub subtitle: Option<Vec<String>>,
+    #[sea_orm(default = "false")]
+    pub deleted: bool,
+    pub source: Option<String>,
+    pub extra: EpisodeExtra,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::bangumi::Entity",
+        from = "Column::BangumiId",
+        to = "super::bangumi::Column::Id"
+    )]
+    Bangumi,
+    #[sea_orm(
+        belongs_to = "super::downloads::Entity",
+        from = "Column::DownloadId",
+        to = "super::downloads::Column::Id"
+    )]
+    Downloads,
+    #[sea_orm(
+        belongs_to = "super::subscriptions::Entity",
+        from = "Column::SubscriptionId",
+        to = "super::subscriptions::Column::Id"
+    )]
+    Subscriptions,
+    #[sea_orm(
+        belongs_to = "super::subscribers::Entity",
+        from = "Column::SubscriberId",
+        to = "super::subscribers::Column::Id"
+    )]
+    Subscriber,
+}
+
+impl Related<super::bangumi::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Bangumi.def()
+    }
+}
+
+impl Related<super::downloads::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Downloads.def()
+    }
+}
+
+impl Related<super::subscriptions::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscriptions.def()
+    }
+}
+
+impl Related<super::subscribers::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscriber.def()
+    }
+}

apps/recorder/src/models/entities/mod.rs

@@ -0,0 +1,7 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.4
+pub mod bangumi;
+pub mod downloads;
+pub mod episodes;
+pub mod subscribers;
+pub mod subscriptions;
+pub mod downloaders;

apps/recorder/src/models/entities/subscribers.rs

@@ -0,0 +1,63 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+
+use sea_orm::{entity::prelude::*, FromJsonQueryResult};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, FromJsonQueryResult)]
+pub struct SubscriberBangumiConfig {
+    pub leading_group_tag: Option<bool>,
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "subscribers")]
+pub struct Model {
+    pub created_at: DateTime,
+    pub updated_at: DateTime,
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    #[sea_orm(unique)]
+    pub pid: String,
+    pub display_name: String,
+    pub downloader_id: Option<i32>,
+    pub bangumi_conf: Option<SubscriberBangumiConfig>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(has_many = "super::subscriptions::Entity")]
+    Subscription,
+    #[sea_orm(
+        belongs_to = "super::downloaders::Entity",
+        from = "Column::DownloaderId",
+        to = "super::downloaders::Column::Id"
+    )]
+    Downloader,
+    #[sea_orm(has_many = "super::bangumi::Entity")]
+    Bangumi,
+    #[sea_orm(has_many = "super::episodes::Entity")]
+    Episode,
+}
+
+impl Related<super::subscriptions::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscription.def()
+    }
+}
+
+impl Related<super::downloaders::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Downloader.def()
+    }
+}
+
+impl Related<super::bangumi::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Bangumi.def()
+    }
+}
+
+impl Related<super::episodes::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Episode.def()
+    }
+}

apps/recorder/src/models/entities/subscriptions.rs

@@ -0,0 +1,66 @@
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(
+    Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, Serialize, Deserialize, DeriveDisplay,
+)]
+#[sea_orm(
+    rs_type = "String",
+    db_type = "Enum",
+    enum_name = "subscription_category"
+)]
+#[serde(rename_all = "snake_case")]
+pub enum SubscriptionCategory {
+    #[sea_orm(string_value = "mikan")]
+    Mikan,
+    #[sea_orm(string_value = "manual")]
+    Manual,
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "subscriptions")]
+pub struct Model {
+    #[sea_orm(column_type = "Timestamp")]
+    pub created_at: DateTime,
+    #[sea_orm(column_type = "Timestamp")]
+    pub updated_at: DateTime,
+    #[sea_orm(primary_key)]
+    pub id: i32,
+    pub display_name: String,
+    pub subscriber_id: i32,
+    pub category: SubscriptionCategory,
+    pub source_url: String,
+    pub enabled: bool,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::subscribers::Entity",
+        from = "Column::SubscriberId",
+        to = "super::subscribers::Column::Id"
+    )]
+    Subscriber,
+    #[sea_orm(has_many = "super::bangumi::Entity")]
+    Bangumi,
+    #[sea_orm(has_many = "super::episodes::Entity")]
+    Episodes,
+}
+
+impl Related<super::subscribers::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscriber.def()
+    }
+}
+
+impl Related<super::bangumi::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Bangumi.def()
+    }
+}
+
+impl Related<super::episodes::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Episodes.def()
+    }
+}

apps/recorder/src/models/episodes.rs

@@ -0,0 +1,104 @@
+use std::sync::Arc;
+
+use loco_rs::app::AppContext;
+use sea_orm::{entity::prelude::*, sea_query::OnConflict, ActiveValue};
+
+use super::bangumi;
+pub use super::entities::episodes::*;
+use crate::{
+    app::AppContextExt,
+    extract::{
+        mikan::{build_mikan_episode_homepage, MikanEpisodeMeta},
+        rawname::parse_episode_meta_from_raw_name,
+    },
+};
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct MikanEpsiodeCreation {
+    pub episode: MikanEpisodeMeta,
+    pub bangumi: Arc<bangumi::Model>,
+}
+
+impl Model {
+    pub async fn add_episodes(
+        ctx: &AppContext,
+        creations: impl IntoIterator<Item = MikanEpsiodeCreation>,
+    ) -> eyre::Result<()> {
+        let db = &ctx.db;
+        let new_episode_active_modes = creations
+            .into_iter()
+            .map(|cr| ActiveModel::from_mikan_episode_meta(ctx, cr))
+            .inspect(|result| {
+                if let Err(e) = result {
+                    tracing::warn!("Failed to create episode: {:?}", e);
+                }
+            })
+            .flatten();
+
+        Entity::insert_many(new_episode_active_modes)
+            .on_conflict(
+                OnConflict::columns([Column::BangumiId, Column::MikanEpisodeId])
+                    .do_nothing()
+                    .to_owned(),
+            )
+            .on_empty_do_nothing()
+            .exec(db)
+            .await?;
+
+        Ok(())
+    }
+}
+
+impl ActiveModel {
+    pub fn from_mikan_episode_meta(
+        ctx: &AppContext,
+        creation: MikanEpsiodeCreation,
+    ) -> eyre::Result<Self> {
+        let item = creation.episode;
+        let bgm = creation.bangumi;
+        let raw_meta = parse_episode_meta_from_raw_name(&item.episode_title)
+            .inspect_err(|e| {
+                tracing::warn!("Failed to parse episode meta: {:?}", e);
+            })
+            .ok()
+            .unwrap_or_default();
+        let homepage = build_mikan_episode_homepage(
+            ctx.get_mikan_client().base_url(),
+            &item.mikan_episode_id,
+        )?;
+
+        Ok(Self {
+            mikan_episode_id: ActiveValue::Set(Some(item.mikan_episode_id)),
+            raw_name: ActiveValue::Set(item.episode_title.clone()),
+            display_name: ActiveValue::Set(item.episode_title.clone()),
+            bangumi_id: ActiveValue::Set(bgm.id),
+            subscription_id: ActiveValue::Set(bgm.subscription_id),
+            subscriber_id: ActiveValue::Set(bgm.subscriber_id),
+            resolution: ActiveValue::Set(raw_meta.resolution),
+            season: ActiveValue::Set(if raw_meta.season > 0 {
+                raw_meta.season
+            } else {
+                bgm.season
+            }),
+            season_raw: ActiveValue::Set(raw_meta.season_raw.or_else(|| bgm.season_raw.clone())),
+            fansub: ActiveValue::Set(raw_meta.fansub.or_else(|| bgm.fansub.clone())),
+            poster_link: ActiveValue::Set(bgm.poster_link.clone()),
+            episode_index: ActiveValue::Set(raw_meta.episode_index),
+            homepage: ActiveValue::Set(Some(homepage.to_string())),
+            subtitle: ActiveValue::Set(raw_meta.subtitle.map(|s| vec![s])),
+            source: ActiveValue::Set(raw_meta.source),
+            extra: ActiveValue::Set(EpisodeExtra {
+                name_zh: raw_meta.name_zh,
+                name_en: raw_meta.name_en,
+                name_jp: raw_meta.name_jp,
+                s_name_en: raw_meta.name_en_no_season,
+                s_name_jp: raw_meta.name_jp_no_season,
+                s_name_zh: raw_meta.name_zh_no_season,
+            }),
+            ..Default::default()
+        })
+    }
+}
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for ActiveModel {}

apps/recorder/src/models/mod.rs

@@ -0,0 +1,10 @@
+pub mod bangumi;
+pub mod downloaders;
+pub mod downloads;
+pub mod entities;
+pub mod episodes;
+pub mod notifications;
+pub mod prelude;
+pub mod query;
+pub mod subscribers;
+pub mod subscriptions;

apps/recorder/src/models/notifications.rs

@@ -0,0 +1,8 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Notification {
+    season: i32,
+    episode_size: u32,
+    poster_url: Option<String>,
+}

apps/recorder/src/models/prelude.rs

@@ -0,0 +1,8 @@
+pub use super::{
+    bangumi::{self, Entity as Bangumi},
+    downloaders::{self, DownloaderCategory, Entity as Downloader},
+    downloads::{self, DownloadMime, DownloadStatus, Entity as Download},
+    episodes::{self, Entity as Episode},
+    subscribers::{self, Entity as Subscriber},
+    subscriptions::{self, Entity as Subscription, SubscriptionCategory},
+};

apps/recorder/src/models/query/mod.rs

@@ -0,0 +1,26 @@
+use sea_orm::{
+    prelude::Expr,
+    sea_query::{Alias, IntoColumnRef, IntoTableRef, Query, SelectStatement},
+    Value,
+};
+
+pub fn filter_values_in<
+    I: IntoIterator<Item = T>,
+    T: Into<Value>,
+    R: IntoTableRef,
+    C: IntoColumnRef + Copy,
+>(
+    tbl_ref: R,
+    col_ref: C,
+    values: I,
+) -> SelectStatement {
+    Query::select()
+        .expr(Expr::col((Alias::new("t"), Alias::new("column1"))))
+        .from_values(values, Alias::new("t"))
+        .left_join(
+            tbl_ref,
+            Expr::col((Alias::new("t"), Alias::new("column1"))).equals(col_ref),
+        )
+        .and_where(Expr::col(col_ref).is_not_null())
+        .to_owned()
+}

apps/recorder/src/models/subscribers.rs

@@ -0,0 +1,97 @@
+use loco_rs::{
+    app::AppContext,
+    model::{ModelError, ModelResult},
+};
+use sea_orm::{entity::prelude::*, ActiveValue, TransactionTrait};
+use serde::{Deserialize, Serialize};
+
+pub use super::entities::subscribers::*;
+
+pub const ROOT_SUBSCRIBER: &str = "konobangu";
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct SubscriberIdParams {
+    pub id: String,
+}
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for ActiveModel {
+    async fn before_save<C>(self, _db: &C, insert: bool) -> Result<Self, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        if insert {
+            let mut this = self;
+            this.pid = ActiveValue::Set(Uuid::new_v4().to_string());
+            Ok(this)
+        } else {
+            Ok(self)
+        }
+    }
+}
+
+impl Model {
+    /// finds a user by the provided pid
+    ///
+    /// # Errors
+    ///
+    /// When could not find user  or DB query error
+    pub async fn find_by_pid(ctx: &AppContext, pid: &str) -> ModelResult<Self> {
+        let db = &ctx.db;
+        let parse_uuid = Uuid::parse_str(pid).map_err(|e| ModelError::Any(e.into()))?;
+        let subscriber = Entity::find()
+            .filter(Column::Pid.eq(parse_uuid))
+            .one(db)
+            .await?;
+        subscriber.ok_or_else(|| ModelError::EntityNotFound)
+    }
+
+    pub async fn find_by_id(ctx: &AppContext, id: i32) -> ModelResult<Self> {
+        let db = &ctx.db;
+
+        let subscriber = Entity::find_by_id(id).one(db).await?;
+        subscriber.ok_or_else(|| ModelError::EntityNotFound)
+    }
+
+    pub async fn find_pid_by_id_with_cache(ctx: &AppContext, id: i32) -> eyre::Result<String> {
+        let db = &ctx.db;
+        let cache = &ctx.cache;
+        let pid = cache
+            .get_or_insert(&format!("subscriber-id2pid::{}", id), async {
+                let subscriber = Entity::find_by_id(id)
+                    .one(db)
+                    .await?
+                    .ok_or_else(|| loco_rs::Error::string(&format!("No such pid for id {}", id)))?;
+                Ok(subscriber.pid)
+            })
+            .await?;
+        Ok(pid)
+    }
+
+    pub async fn find_root(ctx: &AppContext) -> ModelResult<Self> {
+        Self::find_by_pid(ctx, ROOT_SUBSCRIBER).await
+    }
+
+    /// Asynchronously creates a user with a password and saves it to the
+    /// database.
+    ///
+    /// # Errors
+    ///
+    /// When could not save the user into the DB
+    pub async fn create_root(ctx: &AppContext) -> ModelResult<Self> {
+        let db = &ctx.db;
+        let txn = db.begin().await?;
+
+        let user = ActiveModel {
+            display_name: ActiveValue::set(ROOT_SUBSCRIBER.to_string()),
+            pid: ActiveValue::set(ROOT_SUBSCRIBER.to_string()),
+            ..Default::default()
+        }
+        .insert(&txn)
+        .await?;
+
+        txn.commit().await?;
+
+        Ok(user)
+    }
+}

apps/recorder/src/models/subscriptions.rs

@@ -0,0 +1,228 @@
+use std::{collections::HashSet, sync::Arc};
+
+use itertools::Itertools;
+use loco_rs::app::AppContext;
+use sea_orm::{entity::prelude::*, ActiveValue};
+use serde::{Deserialize, Serialize};
+
+pub use super::entities::subscriptions::{self, *};
+use super::{bangumi, episodes, query::filter_values_in};
+use crate::{
+    app::AppContextExt,
+    extract::{
+        mikan::{
+            build_mikan_bangumi_homepage, build_mikan_bangumi_rss_link,
+            parse_mikan_bangumi_meta_from_mikan_homepage,
+            parse_mikan_episode_meta_from_mikan_homepage, parse_mikan_rss_channel_from_rss_link,
+            web_parser::{
+                parse_mikan_bangumi_poster_from_origin_poster_src_with_cache,
+                MikanBangumiPosterMeta,
+            },
+        },
+        rawname::extract_season_from_title_body,
+    },
+    models::episodes::MikanEpsiodeCreation,
+};
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct SubscriptionCreateFromRssDto {
+    pub rss_link: String,
+    pub display_name: String,
+    pub enabled: Option<bool>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "category")]
+pub enum SubscriptionCreateDto {
+    Mikan(SubscriptionCreateFromRssDto),
+}
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for ActiveModel {}
+
+impl ActiveModel {
+    pub fn from_create_dto(create_dto: SubscriptionCreateDto, subscriber_id: i32) -> Self {
+        match create_dto {
+            SubscriptionCreateDto::Mikan(create_dto) => {
+                Self::from_rss_create_dto(SubscriptionCategory::Mikan, create_dto, subscriber_id)
+            }
+        }
+    }
+
+    fn from_rss_create_dto(
+        category: SubscriptionCategory,
+        create_dto: SubscriptionCreateFromRssDto,
+        subscriber_id: i32,
+    ) -> Self {
+        Self {
+            display_name: ActiveValue::Set(create_dto.display_name),
+            enabled: ActiveValue::Set(create_dto.enabled.unwrap_or(false)),
+            subscriber_id: ActiveValue::Set(subscriber_id),
+            category: ActiveValue::Set(category),
+            source_url: ActiveValue::Set(create_dto.rss_link),
+            ..Default::default()
+        }
+    }
+}
+
+impl Model {
+    pub async fn add_subscription(
+        ctx: &AppContext,
+        create_dto: SubscriptionCreateDto,
+        subscriber_id: i32,
+    ) -> eyre::Result<Self> {
+        let db = &ctx.db;
+        let subscription = ActiveModel::from_create_dto(create_dto, subscriber_id);
+
+        Ok(subscription.insert(db).await?)
+    }
+
+    pub async fn toggle_iters(
+        ctx: &AppContext,
+        ids: impl Iterator<Item = i32>,
+        enabled: bool,
+    ) -> eyre::Result<()> {
+        let db = &ctx.db;
+        Entity::update_many()
+            .col_expr(Column::Enabled, Expr::value(enabled))
+            .filter(Column::Id.is_in(ids))
+            .exec(db)
+            .await?;
+        Ok(())
+    }
+
+    pub async fn delete_iters(
+        ctx: &AppContext,
+        ids: impl Iterator<Item = i32>,
+    ) -> eyre::Result<()> {
+        let db = &ctx.db;
+        Entity::delete_many()
+            .filter(Column::Id.is_in(ids))
+            .exec(db)
+            .await?;
+        Ok(())
+    }
+
+    pub async fn pull_subscription(&self, ctx: &AppContext) -> eyre::Result<()> {
+        match &self.category {
+            SubscriptionCategory::Mikan => {
+                let mikan_client = ctx.get_mikan_client();
+                let channel =
+                    parse_mikan_rss_channel_from_rss_link(Some(mikan_client), &self.source_url)
+                        .await?;
+
+                let items = channel.into_items();
+
+                let db = &ctx.db;
+                let items = items.into_iter().collect_vec();
+
+                let mut stmt = filter_values_in(
+                    episodes::Entity,
+                    episodes::Column::MikanEpisodeId,
+                    items
+                        .iter()
+                        .map(|s| Value::from(s.mikan_episode_id.clone())),
+                );
+                stmt.and_where(Expr::col(episodes::Column::SubscriberId).eq(self.subscriber_id));
+
+                let builder = &db.get_database_backend();
+
+                let old_rss_item_mikan_episode_ids_set = db
+                    .query_all(builder.build(&stmt))
+                    .await?
+                    .into_iter()
+                    .flat_map(|qs| qs.try_get_by_index(0))
+                    .collect::<HashSet<String>>();
+
+                let new_rss_items = items
+                    .into_iter()
+                    .filter(|item| {
+                        !old_rss_item_mikan_episode_ids_set.contains(&item.mikan_episode_id)
+                    })
+                    .collect_vec();
+
+                let mut new_metas = vec![];
+                for new_rss_item in new_rss_items.iter() {
+                    new_metas.push(
+                        parse_mikan_episode_meta_from_mikan_homepage(
+                            Some(mikan_client),
+                            new_rss_item.homepage.clone(),
+                        )
+                        .await?,
+                    );
+                }
+
+                let new_mikan_bangumi_groups = new_metas
+                    .into_iter()
+                    .into_group_map_by(|s| (s.mikan_bangumi_id.clone(), s.mikan_fansub_id.clone()));
+
+                for ((mikan_bangumi_id, mikan_fansub_id), new_ep_metas) in new_mikan_bangumi_groups
+                {
+                    let mikan_base_url = ctx.get_mikan_client().base_url();
+                    let bgm_homepage = build_mikan_bangumi_homepage(
+                        mikan_base_url,
+                        &mikan_bangumi_id,
+                        Some(&mikan_fansub_id),
+                    )?;
+                    let bgm_rss_link = build_mikan_bangumi_rss_link(
+                        mikan_base_url,
+                        &mikan_bangumi_id,
+                        Some(&mikan_fansub_id),
+                    )?;
+                    let bgm = Arc::new(
+                        bangumi::Model::get_or_insert_from_mikan(
+                            ctx,
+                            self.subscriber_id,
+                            self.id,
+                            mikan_bangumi_id.to_string(),
+                            mikan_fansub_id.to_string(),
+                            async |am| -> eyre::Result<()> {
+                                let bgm_meta = parse_mikan_bangumi_meta_from_mikan_homepage(
+                                    Some(mikan_client),
+                                    bgm_homepage.clone(),
+                                )
+                                .await?;
+                                let bgm_name = bgm_meta.bangumi_title;
+                                let (_, bgm_season_raw, bgm_season) =
+                                    extract_season_from_title_body(&bgm_name);
+                                am.raw_name = ActiveValue::Set(bgm_name.clone());
+                                am.display_name = ActiveValue::Set(bgm_name);
+                                am.season = ActiveValue::Set(bgm_season);
+                                am.season_raw = ActiveValue::Set(bgm_season_raw);
+                                am.rss_link = ActiveValue::Set(Some(bgm_rss_link.to_string()));
+                                am.homepage = ActiveValue::Set(Some(bgm_homepage.to_string()));
+                                am.fansub = ActiveValue::Set(bgm_meta.fansub);
+                                if let Some(origin_poster_src) = bgm_meta.origin_poster_src {
+                                    if let MikanBangumiPosterMeta {
+                                    poster_src: Some(poster_src),
+                                    ..
+                                    } = parse_mikan_bangumi_poster_from_origin_poster_src_with_cache(
+                                        ctx,
+                                        origin_poster_src,
+                                        self.subscriber_id,
+                                    )
+                                    .await?
+                                    {
+                                        am.poster_link = ActiveValue::Set(Some(poster_src))
+                                    }
+                                }
+                            Ok(())
+                            },
+                        )
+                        .await?,
+                    );
+                    episodes::Model::add_episodes(
+                        ctx,
+                        new_ep_metas.into_iter().map(|item| MikanEpsiodeCreation {
+                            episode: item,
+                            bangumi: bgm.clone(),
+                        }),
+                    )
+                    .await?;
+                }
+                Ok(())
+            }
+            _ => todo!(),
+        }
+    }
+}