feat: init

crates/recorder/.devcontainer/Dockerfile

@@ -0,0 +1,8 @@
+FROM mcr.microsoft.com/vscode/devcontainers/rust:0-1
+
+RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
+     && apt-get -y install --no-install-recommends postgresql-client \
+     && cargo install sea-orm-cli cargo-insta \
+     && chown -R vscode /usr/local/cargo
+
+COPY .env /.env

crates/recorder/.devcontainer/devcontainer.json

@@ -0,0 +1,9 @@
+{
+    "name": "Konobangu Recorder",
+    "dockerComposeFile": "docker-compose.yml",
+    "service": "app",
+    "workspaceFolder": "/workspaces/${localWorkspaceFolderBasename}",
+    "forwardPorts": [
+        3001
+    ]
+}

crates/recorder/.devcontainer/docker-compose.yml

@@ -0,0 +1,40 @@
+version: "3"
+
+services:
+  app:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    command: sleep infinity
+    networks:
+      - db
+      - redis
+    volumes:
+      - ../..:/workspaces:cached
+    env_file:
+      - .env
+  db:
+    image: postgres:15.3-alpine
+    restart: unless-stopped
+    ports:
+      - 5432:5432
+    networks:
+      - db
+    volumes:
+      - postgres-data:/var/lib/postgresql/data
+    env_file:
+      - .env
+  redis:
+    image: redis:latest
+    restart: unless-stopped
+    ports:
+      - 6379:6379
+    networks:
+      - redis
+
+volumes:
+  postgres-data:
+
+networks:
+  db:
+  redis:

crates/recorder/.github/workflows/ci.yaml

@@ -0,0 +1,107 @@
+name: CI
+on:
+  push:
+    branches:
+      - master
+      - main
+  pull_request:
+
+env:
+  RUST_TOOLCHAIN: stable
+  TOOLCHAIN_PROFILE: minimal
+
+jobs:
+  rustfmt:
+    name: Check Style
+    runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+
+    steps:
+      - name: Checkout the code
+        uses: actions/checkout@v4
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: ${{ env.TOOLCHAIN_PROFILE }}
+          toolchain: ${{ env.RUST_TOOLCHAIN }}
+          override: true
+          components: rustfmt
+      - name: Run cargo fmt
+        uses: actions-rs/cargo@v1
+        with:
+          command: fmt
+          args: --all -- --check
+
+  clippy:
+    name: Run Clippy
+    runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+
+    steps:
+      - name: Checkout the code
+        uses: actions/checkout@v4
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: ${{ env.TOOLCHAIN_PROFILE }}
+          toolchain: ${{ env.RUST_TOOLCHAIN }}
+          override: true
+      - name: Setup Rust cache
+        uses: Swatinem/rust-cache@v2
+      - name: Run cargo clippy
+        uses: actions-rs/cargo@v1
+        with:
+          command: clippy
+          args: --all-features -- -D warnings -W clippy::pedantic -W clippy::nursery -W rust-2018-idioms
+
+  test:
+    name: Run Tests
+    runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+
+    services:
+      redis:
+        image: redis
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - "6379:6379"
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_DB: postgress_test
+          POSTGRES_USER: postgress
+          POSTGRES_PASSWORD: postgress
+        ports:
+          - "5432:5432"
+        # Set health checks to wait until postgres has started
+        options: --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - name: Checkout the code
+        uses: actions/checkout@v4
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: ${{ env.TOOLCHAIN_PROFILE }}
+          toolchain: ${{ env.RUST_TOOLCHAIN }}
+          override: true
+      - name: Setup Rust cache
+        uses: Swatinem/rust-cache@v2
+      - name: Run cargo test
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --all-features --all
+        env:
+          REDIS_URL: redis://localhost:${{job.services.redis.ports[6379]}}
+          DATABASE_URL: postgres://postgress:postgress@localhost:5432/postgress_test

crates/recorder/.gitignore

@@ -0,0 +1,17 @@
+**/config/local.yaml
+**/config/*.local.yaml
+
+# Generated by Cargo
+# will have compiled files and executables
+debug/
+target/
+
+# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
+# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html
+Cargo.lock
+
+# These are backup files generated by rustfmt
+**/*.rs.bk
+
+# MSVC Windows builds of rustc generate these, which store debugging information
+*.pdb

crates/recorder/Cargo.toml

@@ -0,0 +1,50 @@
+[package]
+name = "recorder"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+
+loco-rs = { version = "0.3.1" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+eyre = "0.6"
+tokio = { version = "1.33.0", default-features = false }
+async-trait = "0.1.74"
+tracing = "0.1.40"
+chrono = "0.4"
+validator = { version = "0.16" }
+sea-orm = { version = "1.0.0-rc.1", features = [
+  "sqlx-sqlite",
+  "sqlx-postgres",
+  "runtime-tokio-rustls",
+  "macros",
+] }
+
+axum = "0.7.1"
+include_dir = "0.7"
+uuid = { version = "1.6.0", features = ["v4"] }
+tracing-subscriber = { version = "0.3.17", features = ["env-filter", "json"] }
+sea-orm-migration = { version = "1.0.0-rc.1", features = [
+  "runtime-tokio-rustls",
+] }
+reqwest = "0.11.24"
+thiserror = "1.0.57"
+rss = "2.0.7"
+
+[lib]
+name = "recorder"
+path = "src/lib.rs"
+
+[[bin]]
+name = "recorder_cli"
+path = "src/bin/main.rs"
+required-features = []
+
+[dev-dependencies]
+serial_test = "2.0.0"
+rstest = "0.18.2"
+loco-rs = { version = "0.3.1", features = ["testing"] }
+insta = { version = "1.34.0", features = ["redactions", "yaml", "filters"] }

crates/recorder/examples/playground.rs

@@ -0,0 +1,33 @@
+use eyre::Context;
+#[allow(unused_imports)]
+use loco_rs::{cli::playground, prelude::*};
+use recorder::app::App;
+
+async fn fetch_and_parse_rss_demo () -> eyre::Result<()> {
+    let url =
+    "https://mikanani.me/RSS/MyBangumi?token=FE9tccsML2nBPUUqpCuJW2uJZydAXCntHJ7RpD9LDP8%3d";
+
+    let res = reqwest::get(url).await?.bytes().await?;
+    let channel = rss::Channel::read_from(&res[..])?;
+    println!("channel: {:#?}", channel);
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> eyre::Result<()> {
+
+    fetch_and_parse_rss_demo().await?;
+
+    // let active_model: articles::ActiveModel = ActiveModel {
+    //     title: Set(Some("how to build apps in 3 steps".to_string())),
+    //     content: Set(Some("use Loco: https://loco.rs".to_string())),
+    //     ..Default::default()
+    // };
+    // active_model.insert(&ctx.db).await.unwrap();
+
+    // let res = articles::Entity::find().all(&ctx.db).await.unwrap();
+    // println!("{:?}", res);
+    println!("welcome to playground. edit me at `examples/playground.rs`");
+
+    Ok(())
+}

crates/recorder/src/app.rs

@@ -0,0 +1,62 @@
+use std::path::Path;
+
+use async_trait::async_trait;
+use loco_rs::{
+    app::{AppContext, Hooks},
+    boot::{create_app, BootResult, StartMode},
+    controller::AppRoutes,
+    db::truncate_table,
+    environment::Environment,
+    task::Tasks,
+    worker::{AppWorker, Processor},
+    Result,
+};
+use sea_orm::DatabaseConnection;
+
+use crate::{
+    controllers, migrations::Migrator, models::_entities::subscribers,
+    workers::downloader::DownloadWorker,
+};
+
+pub struct App;
+#[async_trait]
+impl Hooks for App {
+    fn app_name() -> &'static str {
+        env!("CARGO_CRATE_NAME")
+    }
+
+    fn app_version() -> String {
+        format!(
+            "{} ({})",
+            env!("CARGO_PKG_VERSION"),
+            option_env!("BUILD_SHA")
+                .or(option_env!("GITHUB_SHA"))
+                .unwrap_or("dev")
+        )
+    }
+
+    async fn boot(mode: StartMode, environment: &Environment) -> Result<BootResult> {
+        create_app::<Self, Migrator>(mode, environment).await
+    }
+
+    fn routes(_ctx: &AppContext) -> AppRoutes {
+        AppRoutes::with_default_routes()
+            .prefix("/api")
+            .add_route(controllers::subscribers::routes())
+    }
+
+    fn connect_workers<'a>(p: &'a mut Processor, ctx: &'a AppContext) {
+        p.register(DownloadWorker::build(ctx));
+    }
+
+    fn register_tasks(_tasks: &mut Tasks) {}
+
+    async fn truncate(db: &DatabaseConnection) -> Result<()> {
+        truncate_table(db, subscribers::Entity).await?;
+        Ok(())
+    }
+
+    async fn seed(_db: &DatabaseConnection, _base: &Path) -> Result<()> {
+        Ok(())
+    }
+}

crates/recorder/src/bin/main.rs

@@ -0,0 +1,8 @@
+use loco_rs::cli;
+use recorder::migrations::Migrator;
+use recorder::app::App;
+
+#[tokio::main]
+async fn main() -> eyre::Result<()> {
+    cli::main::<App, Migrator>().await
+}

crates/recorder/src/controllers/mod.rs

@@ -0,0 +1 @@
+pub mod subscribers;

crates/recorder/src/controllers/subscribers.rs

@@ -0,0 +1,14 @@
+use loco_rs::prelude::*;
+
+use crate::{models::_entities::subscribers, views::subscribers::CurrentResponse};
+
+async fn current(State(ctx): State<AppContext>) -> Result<Json<CurrentResponse>> {
+    let subscriber = subscribers::Model::find_root(&ctx.db).await?;
+    format::json(CurrentResponse::new(&subscriber))
+}
+
+pub fn routes() -> Routes {
+    Routes::new()
+        .prefix("subscribers")
+        .add("/current", get(current))
+}

crates/recorder/src/downloader/mod.rs

@@ -0,0 +1,2 @@
+pub mod aria;
+pub mod qbitorrent;

crates/recorder/src/lib.rs

@@ -0,0 +1,10 @@
+pub mod app;
+pub mod controllers;
+pub mod downloader;
+pub mod migrations;
+pub mod models;
+pub mod rss;
+pub mod subscriptions;
+pub mod tasks;
+pub mod views;
+pub mod workers;

crates/recorder/src/migrations/defs.rs

@@ -0,0 +1,40 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(Iden)]
+pub enum Subscribers {
+    Table,
+    Id,
+    Pid,
+    DisplayName,
+}
+
+#[derive(Iden)]
+pub enum Subscriptions {
+    Table,
+    Id,
+    SubscriberId,
+    DisplayName,
+    Category,
+    SourceUrl,
+    Aggregate,
+    Enabled,
+}
+
+#[derive(Iden)]
+pub enum Bangumi {
+    Table,
+    Id,
+    DisplayName,
+    SubscriptionId,
+}
+
+#[derive(Iden)]
+pub enum Episodes {
+    Table,
+    Id,
+    DisplayName,
+    BangumiId,
+    DownloadUrl,
+    DownloadProgress,
+    OutputName,
+}

crates/recorder/src/migrations/m20220101_000001_init.rs

@@ -0,0 +1,125 @@
+use sea_orm::sea_query::extension::postgres::Type;
+use sea_orm_migration::{prelude::*, schema::*};
+
+use super::defs::{Bangumi, Episodes, Subscribers, Subscriptions};
+use crate::models::subscribers::ROOT_SUBSCRIBER;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                table_auto(Subscribers::Table)
+                    .col(pk_auto(Subscribers::Id))
+                    .col(string_len_uniq(Subscribers::Pid, 64))
+                    .col(string(Subscribers::DisplayName))
+                    .to_owned(),
+            )
+            .await?;
+
+        let insert = Query::insert()
+            .into_table(Subscribers::Table)
+            .columns([Subscribers::Pid, Subscribers::DisplayName])
+            .values_panic([ROOT_SUBSCRIBER.into(), ROOT_SUBSCRIBER.into()])
+            .to_owned();
+        manager.exec_stmt(insert).await?;
+
+        manager
+            .create_type(
+                Type::create()
+                    .as_enum(Alias::new("subscription_category"))
+                    .values([
+                        Alias::new("mikan"),
+                        Alias::new("manual"),
+                        Alias::new("bangumi"),
+                    ])
+                    .to_owned(),
+            )
+            .await?;
+
+        manager
+            .create_table(
+                table_auto(Subscriptions::Table)
+                    .col(pk_auto(Subscriptions::Id))
+                    .col(string(Subscriptions::DisplayName))
+                    .col(integer(Subscriptions::SubscriberId))
+                    .col(text(Subscriptions::SourceUrl))
+                    .col(boolean(Subscriptions::Aggregate))
+                    .col(boolean(Subscriptions::Enabled))
+                    .foreign_key(
+                        ForeignKey::create()
+                            .name("subscription_subscriber_id")
+                            .from(Subscriptions::Table, Subscriptions::SubscriberId)
+                            .to(Subscribers::Table, Subscribers::Id),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        manager
+            .create_table(
+                table_auto(Bangumi::Table)
+                    .col(pk_auto(Bangumi::Id))
+                    .col(text(Bangumi::DisplayName))
+                    .col(integer(Bangumi::SubscriptionId))
+                    .foreign_key(
+                        ForeignKey::create()
+                            .name("bangumi_subscription_id")
+                            .from(Bangumi::Table, Bangumi::SubscriptionId)
+                            .to(Subscriptions::Table, Subscriptions::Id),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        manager
+            .create_table(
+                table_auto(Episodes::Table)
+                    .col(pk_auto(Episodes::Id))
+                    .col(text(Episodes::DisplayName))
+                    .col(integer(Episodes::BangumiId))
+                    .col(text(Episodes::DownloadUrl))
+                    .col(tiny_integer(Episodes::DownloadProgress).default(0))
+                    .col(text(Episodes::OutputName))
+                    .foreign_key(
+                        ForeignKey::create()
+                            .name("episode_bangumi_id")
+                            .from(Episodes::Table, Episodes::BangumiId)
+                            .to(Bangumi::Table, Bangumi::Id),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_table(Table::drop().table(Episodes::Table).to_owned())
+            .await?;
+
+        manager
+            .drop_table(Table::drop().table(Bangumi::Table).to_owned())
+            .await?;
+
+        manager
+            .drop_table(Table::drop().table(Subscriptions::Table).to_owned())
+            .await?;
+
+        manager
+            .drop_type(
+                Type::drop()
+                    .name(Alias::new("subscription_category"))
+                    .to_owned(),
+            )
+            .await?;
+
+        manager
+            .drop_table(Table::drop().table(Subscribers::Table).to_owned())
+            .await
+    }
+}

crates/recorder/src/migrations/mod.rs

@@ -0,0 +1,13 @@
+pub use sea_orm_migration::prelude::*;
+
+pub mod defs;
+pub mod m20220101_000001_init;
+
+pub struct Migrator;
+
+#[async_trait::async_trait]
+impl MigratorTrait for Migrator {
+    fn migrations() -> Vec<Box<dyn MigrationTrait>> {
+        vec![Box::new(m20220101_000001_init::Migration)]
+    }
+}

crates/recorder/src/models/_entities/bangumi.rs

@@ -0,0 +1,39 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[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 display_name: String,
+    pub subscription_id: i32,
+}
+
+#[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::episodes::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Episode.def()
+    }
+}
+
+impl Related<super::subscriptions::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscription.def()
+    }
+}

crates/recorder/src/models/_entities/episodes.rs

@@ -0,0 +1,34 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[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,
+    pub display_name: String,
+    pub bangumi_id: i32,
+    pub download_url: String,
+    pub download_progress: i32,
+    pub output_name: String,
+}
+
+#[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,
+}
+
+impl Related<super::bangumi::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Bangumi.def()
+    }
+}

crates/recorder/src/models/_entities/mod.rs

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

crates/recorder/src/models/_entities/prelude.rs

@@ -0,0 +1,6 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.4
+
+pub use super::{
+    bangumi::Entity as Bangumi, episodes::Entity as Episodes, subscribers::Entity as Subscribers,
+    subscriptions::Entity as Subscriptions,
+};

crates/recorder/src/models/_entities/subscribers.rs

@@ -0,0 +1,28 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[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,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(has_many = "super::subscriptions::Entity")]
+    Subscription,
+}
+
+impl Related<super::subscriptions::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Subscription.def()
+    }
+}

crates/recorder/src/models/_entities/subscriptions.rs

@@ -0,0 +1,59 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.12.2
+
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
+#[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,
+    #[sea_orm(string_value = "bangumi")]
+    Bangumi,
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
+#[sea_orm(table_name = "subscriptions")]
+pub struct Model {
+    pub created_at: DateTime,
+    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 aggregate: bool,
+    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,
+}
+
+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()
+    }
+}

crates/recorder/src/models/bangumi.rs

@@ -0,0 +1,6 @@
+use sea_orm::entity::prelude::*;
+
+pub use super::_entities::bangumi::{self, ActiveModel, Entity, Model};
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for super::_entities::bangumi::ActiveModel {}

crates/recorder/src/models/episodes.rs

@@ -0,0 +1,6 @@
+use sea_orm::entity::prelude::*;
+
+pub use super::_entities::episodes::{self, ActiveModel, Entity, Model};
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for super::_entities::episodes::ActiveModel {}

crates/recorder/src/models/mod.rs

@@ -0,0 +1,5 @@
+pub mod _entities;
+pub mod bangumi;
+pub mod episodes;
+pub mod subscribers;
+pub mod subscriptions;

crates/recorder/src/models/subscribers.rs

@@ -0,0 +1,70 @@
+use loco_rs::model::{ModelError, ModelResult};
+use sea_orm::{entity::prelude::*, ActiveValue, TransactionTrait};
+use serde::{Deserialize, Serialize};
+
+pub use super::_entities::subscribers::{self, ActiveModel, Entity, Model};
+
+pub const ROOT_SUBSCRIBER: &str = "konobangu";
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct SubscriberIdParams {
+    pub id: String,
+}
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for super::_entities::subscribers::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 super::_entities::subscribers::Model {
+    /// finds a user by the provided pid
+    ///
+    /// # Errors
+    ///
+    /// When could not find user  or DB query error
+    pub async fn find_by_pid(db: &DatabaseConnection, pid: &str) -> ModelResult<Self> {
+        let parse_uuid = Uuid::parse_str(pid).map_err(|e| ModelError::Any(e.into()))?;
+        let subscriber = subscribers::Entity::find()
+            .filter(subscribers::Column::Pid.eq(parse_uuid))
+            .one(db)
+            .await?;
+        subscriber.ok_or_else(|| ModelError::EntityNotFound)
+    }
+
+    pub async fn find_root(db: &DatabaseConnection) -> ModelResult<Self> {
+        Self::find_by_pid(db, 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(db: &DatabaseConnection) -> ModelResult<Self> {
+        let txn = db.begin().await?;
+
+        let user = subscribers::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)
+    }
+}

crates/recorder/src/models/subscriptions.rs

@@ -0,0 +1,6 @@
+use sea_orm::entity::prelude::*;
+
+pub use super::_entities::subscriptions::{self, ActiveModel, Entity, Model};
+
+#[async_trait::async_trait]
+impl ActiveModelBehavior for super::_entities::subscriptions::ActiveModel {}

crates/recorder/src/rss/engine.rs

@@ -0,0 +1,23 @@
+use serde::{Deserialize, Serialize};
+
+use crate::models::subscriptions::subscriptions;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RssTorrent {}
+
+#[derive(Debug)]
+pub struct RssEngine {}
+
+impl RssEngine {
+    // pub async fn get_rss_torrents(
+    //     rss_subscription: &subscriptions::ActiveModel,
+    // ) -> eyre::Result<Vec<RssTorrent>> {
+    //     Ok(())
+    // }
+
+    pub async fn get_torrents(url: &str) -> eyre::Result<rss::Channel> {
+        let content = reqwest::get(url).await?.bytes().await?;
+        let channel: rss::Channel = rss::Channel::read_from(&content[..])?;
+        Ok(channel)
+    }
+}

crates/recorder/src/rss/mod.rs

@@ -0,0 +1 @@
+pub mod engine;

crates/recorder/src/subscriptions/bangumi.rs

@@ -0,0 +1 @@
+

crates/recorder/src/subscriptions/mikan.rs

@@ -0,0 +1,23 @@
+use crate::rss::engine::RssEngine;
+
+pub struct MikanRssCreateDto {
+    pub rss_link: String,
+    pub display_name: String,
+    pub aggregate: bool,
+    pub enabled: Option<bool>,
+}
+
+pub struct MikanSubscriptionEngine {
+}
+
+impl MikanSubscriptionEngine {
+    pub async fn add_rss(create_dto: MikanRssCreateDto) -> eyre::Result<()> {
+        let content = reqwest::get(&create_dto.rss_link).await?.bytes().await?;
+        let channel = rss::Channel::read_from(&content[..])?;
+
+        Ok(())
+    }
+}
+
+pub struct MikanSubscriptionItem {
+}

crates/recorder/src/subscriptions/mod.rs

@@ -0,0 +1,2 @@
+pub mod bangumi;
+pub mod mikan;

crates/recorder/src/tasks/mod.rs

@@ -0,0 +1 @@
+

crates/recorder/src/views/mod.rs

@@ -0,0 +1 @@
+pub mod subscribers;

crates/recorder/src/views/subscribers.rs

@@ -0,0 +1,19 @@
+use serde::{Deserialize, Serialize};
+
+use crate::models::_entities::subscribers;
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct CurrentResponse {
+    pub pid: String,
+    pub display_name: String,
+}
+
+impl CurrentResponse {
+    #[must_use]
+    pub fn new(user: &subscribers::Model) -> Self {
+        Self {
+            pid: user.pid.to_string(),
+            display_name: user.display_name.to_string(),
+        }
+    }
+}

crates/recorder/src/workers/downloader.rs

@@ -0,0 +1,43 @@
+use std::time::Duration;
+
+use loco_rs::prelude::*;
+use serde::{Deserialize, Serialize};
+use tokio::time::sleep;
+
+use crate::models::subscribers;
+
+pub struct DownloadWorker {
+    pub ctx: AppContext,
+}
+
+#[derive(Deserialize, Debug, Serialize)]
+pub struct DownloadWorkerArgs {
+    pub user_guid: String,
+}
+
+impl worker::AppWorker<DownloadWorkerArgs> for DownloadWorker {
+    fn build(ctx: &AppContext) -> Self {
+        Self { ctx: ctx.clone() }
+    }
+}
+
+#[async_trait]
+impl worker::Worker<DownloadWorkerArgs> for DownloadWorker {
+    async fn perform(&self, args: DownloadWorkerArgs) -> worker::Result<()> {
+        // TODO: Some actual work goes here...
+        println!("================================================");
+        println!("Sending payment report to user {}", args.user_guid);
+
+        sleep(Duration::from_millis(2000)).await;
+
+        let all = subscribers::Entity::find()
+            .all(&self.ctx.db)
+            .await
+            .map_err(Box::from)?;
+        for user in &all {
+            println!("user: {}", user.id);
+        }
+        println!("================================================");
+        Ok(())
+    }
+}

crates/recorder/src/workers/mod.rs

@@ -0,0 +1 @@
+pub mod downloader;

crates/recorder/tests/mod.rs

@@ -0,0 +1,3 @@
+mod models;
+mod requests;
+mod tasks;

crates/recorder/tests/models/mod.rs

@@ -0,0 +1 @@
+mod subscribers;

crates/recorder/tests/models/snapshots/can_find_by_pid@subscribers-2.snap

@@ -0,0 +1,7 @@
+---
+source: tests/models/subscribers.rs
+expression: non_existing_subscriber_results
+---
+Err(
+    EntityNotFound,
+)

crates/recorder/tests/models/snapshots/can_find_by_pid@subscribers.snap

@@ -0,0 +1,13 @@
+---
+source: tests/models/subscribers.rs
+expression: existing_subscriber
+---
+Ok(
+    Model {
+        created_at: 2023-11-12T12:34:56.789,
+        updated_at: 2023-11-12T12:34:56.789,
+        id: 1,
+        pid: "11111111-1111-1111-1111-111111111111",
+        display_name: "user1"
+    },
+)

crates/recorder/tests/models/subscribers.rs

@@ -0,0 +1,27 @@
+use insta::assert_debug_snapshot;
+use loco_rs::testing;
+use recorder::{app::App, models::subscribers::Model};
+use serial_test::serial;
+
+macro_rules! configure_insta {
+    ($($expr:expr),*) => {
+        let mut settings = insta::Settings::clone_current();
+        settings.set_prepend_module_to_snapshot(false);
+        settings.set_snapshot_suffix("users");
+        let _guard = settings.bind_to_scope();
+    };
+}
+
+#[tokio::test]
+#[serial]
+async fn can_find_by_pid() {
+    configure_insta!();
+
+    let boot = testing::boot_test::<App>().await.unwrap();
+    testing::seed::<App>(&boot.app_context.db).await.unwrap();
+
+    let existing_subscriber =
+        Model::find_by_pid(&boot.app_context.db, "11111111-1111-1111-1111-111111111111").await;
+
+    assert_debug_snapshot!(existing_subscriber);
+}

crates/recorder/tests/requests/mod.rs

@@ -0,0 +1,2 @@
+mod notes;
+mod subscribers;

crates/recorder/tests/requests/subscribers.rs

@@ -0,0 +1,32 @@
+use insta::{assert_debug_snapshot, with_settings};
+use loco_rs::testing;
+use recorder::app::App;
+use serial_test::serial;
+
+// TODO: see how to dedup / extract this to app-local test utils
+// not to framework, because that would require a runtime dep on insta
+macro_rules! configure_insta {
+    ($($expr:expr),*) => {
+        let mut settings = insta::Settings::clone_current();
+        settings.set_prepend_module_to_snapshot(false);
+        settings.set_snapshot_suffix("user_request");
+        let _guard = settings.bind_to_scope();
+    };
+}
+
+#[tokio::test]
+#[serial]
+async fn can_get_current_user() {
+    configure_insta!();
+
+    testing::request::<App, _, _>(|request, _ctx| async move {
+        let response = request.get("/api/user/current").await;
+
+        with_settings!({
+            filters => testing::cleanup_user_model()
+        }, {
+            assert_debug_snapshot!((response.status_code(), response.text()));
+        });
+    })
+    .await;
+}

crates/recorder/tests/tasks/mod.rs

@@ -0,0 +1 @@
+pub mod seed;

crates/recorder/tests/tasks/seed.rs

@@ -0,0 +1,43 @@
+//! This task implements data seeding functionality for initializing new
+//! development/demo environments.
+//!
+//! # Example
+//!
+//! Run the task with the following command:
+//! ```sh
+//! cargo run task
+//! ```
+//!
+//! To override existing data and reset the data structure, use the following
+//! command with the `refresh:true` argument:
+//! ```sh
+//! cargo run task seed_data refresh:true
+//! ```
+use std::collections::BTreeMap;
+
+use loco_rs::{db, prelude::*};
+use migration::Migrator;
+use recorder::app::App;
+
+#[allow(clippy::module_name_repetitions)]
+pub struct SeedData;
+#[async_trait]
+impl Task for SeedData {
+    fn task(&self) -> TaskInfo {
+        TaskInfo {
+            name: "seed_data".to_string(),
+            detail: "Task for seeding data".to_string(),
+        }
+    }
+
+    async fn run(&self, app_context: &AppContext, vars: &BTreeMap<String, String>) -> Result<()> {
+        let refresh = vars.get("refresh").is_some_and(|refresh| refresh == "true");
+
+        if refresh {
+            db::reset::<Migrator>(&app_context.db).await?;
+        }
+        let path = std::path::Path::new("src/fixtures");
+        db::run_app_seed::<App>(&app_context.db, path).await?;
+        Ok(())
+    }
+}