dumtruck/konobangu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: support system tasks

Browse Source
This commit is contained in:
master
2025-07-03 03:48:23 +08:00
parent 5b001f9584
commit 1d0aa8d7f1
44 changed files with 1833 additions and 595 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
apps/recorder/Cargo.toml
View File

@@ -6,7 +6,7 @@ edition = "2024"
[features]
default = ["jxl"]
playground = ["dep:inquire", "dep:color-eyre", "dep:polars"]
playground = ["dep:inquire", "dep:color-eyre", "dep:polars", "test-utils"]
testcontainers = [
"dep:testcontainers",
"dep:testcontainers-modules",
@@ -15,6 +15,7 @@ testcontainers = [
"testcontainers-modules/postgres",
]
jxl = ["dep:jpegxl-rs", "dep:jpegxl-sys"]
test-utils = []
[lib]
name = "recorder"

11
apps/recorder/src/app/builder.rs
View File

@@ -131,11 +131,12 @@ impl AppBuilder {
}
pub fn working_dir_from_manifest_dir(self) -> Self {
let manifest_dir = if cfg!(debug_assertions) || cfg!(test) || cfg!(feature = "playground") {
env!("CARGO_MANIFEST_DIR")
} else {
"./apps/recorder"
};
#[cfg(any(test, debug_assertions, feature = "test-utils"))]
let manifest_dir = env!("CARGO_MANIFEST_DIR");
#[cfg(not(any(test, debug_assertions, feature = "test-utils")))]
let manifest_dir = "./apps/recorder";
self.working_dir(manifest_dir.to_string())
}
}

53
apps/recorder/src/extract/mikan/subscription.rs
View File

@@ -546,14 +546,12 @@ impl MikanBangumiSubscription {
#[cfg(test)]
#[allow(unused_variables)]
mod tests {
use std::sync::Arc;
use rstest::{fixture, rstest};
use sea_orm::{ActiveModelTrait, ActiveValue, EntityTrait};
use tracing::Level;
use crate::{
app::AppContextTrait,
errors::RecorderResult,
extract::mikan::{
MikanBangumiHash, MikanSeasonFlowUrlMeta, MikanSeasonStr,
@@ -564,34 +562,11 @@ mod tests {
subscriptions::{self, SubscriptionTrait},
},
test_utils::{
app::{TestingAppContext, TestingAppContextPreset},
mikan::{MikanMockServer, build_testing_mikan_credential_form},
app::TestingPreset, mikan::build_testing_mikan_credential_form,
tracing::try_init_testing_tracing,
},
};
struct TestingResources {
pub app_ctx: Arc<dyn AppContextTrait>,
pub mikan_server: MikanMockServer,
}
async fn build_testing_app_context() -> RecorderResult<TestingResources> {
let mikan_server = MikanMockServer::new().await?;
let mikan_base_url = mikan_server.base_url().clone();
let app_ctx = TestingAppContext::from_preset(TestingAppContextPreset {
mikan_base_url: mikan_base_url.to_string(),
database_config: None,
})
.await?;
Ok(TestingResources {
app_ctx,
mikan_server,
})
}
#[fixture]
fn before_each() {
try_init_testing_tracing(Level::DEBUG);
@@ -600,10 +575,10 @@ mod tests {
#[rstest]
#[tokio::test]
async fn test_mikan_season_subscription_sync_feeds(before_each: ()) -> RecorderResult<()> {
let TestingResources {
app_ctx,
mut mikan_server,
} = build_testing_app_context().await?;
let mut preset = TestingPreset::default().await?;
let app_ctx = preset.app_ctx.clone();
let mikan_server = &mut preset.mikan_server;
let _resources_mock = mikan_server.mock_resources_with_doppel();
@@ -662,10 +637,11 @@ mod tests {
#[rstest]
#[tokio::test]
async fn test_mikan_subscriber_subscription_sync_feeds(before_each: ()) -> RecorderResult<()> {
let TestingResources {
app_ctx,
mut mikan_server,
} = build_testing_app_context().await?;
let mut preset = TestingPreset::default().await?;
let app_ctx = preset.app_ctx.clone();
let mikan_server = &mut preset.mikan_server;
let _resources_mock = mikan_server.mock_resources_with_doppel();
@@ -729,10 +705,11 @@ mod tests {
#[rstest]
#[tokio::test]
async fn test_mikan_bangumi_subscription_sync_feeds(before_each: ()) -> RecorderResult<()> {
let TestingResources {
app_ctx,
mut mikan_server,
} = build_testing_app_context().await?;
let mut preset = TestingPreset::default().await?;
let app_ctx = preset.app_ctx.clone();
let mikan_server = &mut preset.mikan_server;
let _resources_mock = mikan_server.mock_resources_with_doppel();

55
apps/recorder/src/extract/mikan/web.rs
View File

@@ -35,7 +35,7 @@ use crate::{
EncodeWebpOptions,
},
storage::StorageContentCategory,
task::{OptimizeImageTask, SystemTask},
task::OptimizeImageTask,
};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -818,11 +818,14 @@ pub async fn scrape_mikan_poster_meta_from_image_url(
let webp_storage_path = storage_path.with_extension("webp");
if storage_service.exists(&webp_storage_path).await?.is_none() {
task_service
.add_system_task(SystemTask::OptimizeImage(OptimizeImageTask {
source_path: storage_path.clone().to_string(),
target_path: webp_storage_path.to_string(),
format_options: EncodeImageOptions::Webp(EncodeWebpOptions::default()),
}))
.add_system_task(
OptimizeImageTask::builder()
.source_path(storage_path.clone().to_string())
.target_path(webp_storage_path.to_string())
.format_options(EncodeImageOptions::Webp(EncodeWebpOptions::default()))
.build()
.into(),
)
.await?;
}
}
@@ -830,11 +833,14 @@ pub async fn scrape_mikan_poster_meta_from_image_url(
let avif_storage_path = storage_path.with_extension("avif");
if storage_service.exists(&avif_storage_path).await?.is_none() {
task_service
.add_system_task(SystemTask::OptimizeImage(OptimizeImageTask {
source_path: storage_path.clone().to_string(),
target_path: avif_storage_path.to_string(),
format_options: EncodeImageOptions::Avif(EncodeAvifOptions::default()),
}))
.add_system_task(
OptimizeImageTask::builder()
.source_path(storage_path.clone().to_string())
.target_path(avif_storage_path.to_string())
.format_options(EncodeImageOptions::Avif(EncodeAvifOptions::default()))
.build()
.into(),
)
.await?;
}
}
@@ -842,11 +848,14 @@ pub async fn scrape_mikan_poster_meta_from_image_url(
let jxl_storage_path = storage_path.with_extension("jxl");
if storage_service.exists(&jxl_storage_path).await?.is_none() {
task_service
.add_system_task(SystemTask::OptimizeImage(OptimizeImageTask {
source_path: storage_path.clone().to_string(),
target_path: jxl_storage_path.to_string(),
format_options: EncodeImageOptions::Jxl(EncodeJxlOptions::default()),
}))
.add_system_task(
OptimizeImageTask::builder()
.source_path(storage_path.clone().to_string())
.target_path(jxl_storage_path.to_string())
.format_options(EncodeImageOptions::Jxl(EncodeJxlOptions::default()))
.build()
.into(),
)
.await?;
}
}
@@ -1089,7 +1098,7 @@ mod test {
use super::*;
use crate::test_utils::{
app::{TestingAppContext, TestingAppContextPreset},
app::{TestingAppContext, TestingPreset},
crypto::build_testing_crypto_service,
database::build_testing_database_service,
mikan::{
@@ -1137,17 +1146,13 @@ mod test {
#[rstest]
#[tokio::test]
async fn test_scrape_mikan_poster_meta_from_image_url(before_each: ()) -> RecorderResult<()> {
let mut mikan_server = MikanMockServer::new().await?;
let mut preset = TestingPreset::default().await?;
let mikan_base_url = mikan_server.base_url().clone();
let app_ctx = preset.app_ctx.clone();
let app_ctx = TestingAppContext::from_preset(TestingAppContextPreset {
mikan_base_url: mikan_base_url.to_string(),
database_config: None,
})
.await?;
let mikan_base_url = preset.mikan_server.base_url().clone();
let resources_mock = mikan_server.mock_resources_with_doppel();
let resources_mock = preset.mikan_server.mock_resources_with_doppel();
let bangumi_poster_url = mikan_base_url.join("/images/Bangumi/202309/5ce9fed1.jpg")?;

3
apps/recorder/src/graphql/domains/cron.rs
View File

@@ -6,6 +6,7 @@ use crate::{
domains::{
subscriber_tasks::restrict_subscriber_tasks_for_entity,
subscribers::restrict_subscriber_for_entity,
system_tasks::restrict_system_tasks_for_entity,
},
infra::{custom::register_entity_default_writable, name::get_entity_and_column_name},
},
@@ -17,6 +18,7 @@ fn skip_columns_for_entity_input(context: &mut BuilderContext) {
if matches!(
column,
cron::Column::SubscriberTask
| cron::Column::SystemTask
| cron::Column::CronExpr
| cron::Column::Enabled
| cron::Column::TimeoutMs
@@ -44,6 +46,7 @@ pub fn register_cron_to_schema_context(context: &mut BuilderContext) {
restrict_subscriber_for_entity::<cron::Entity>(context, &cron::Column::SubscriberId);
restrict_subscriber_tasks_for_entity::<cron::Entity>(context, &cron::Column::SubscriberTask);
restrict_system_tasks_for_entity::<cron::Entity>(context, &cron::Column::SystemTask);
skip_columns_for_entity_input(context);
}

3
apps/recorder/src/graphql/domains/mod.rs
View File

@@ -1,6 +1,7 @@
pub mod credential_3rd;
pub mod bangumi;
pub mod cron;
pub mod downloaders;
pub mod downloads;
pub mod episodes;
@@ -10,4 +11,4 @@ pub mod subscribers;
pub mod subscription_bangumi;
pub mod subscription_episode;
pub mod subscriptions;
pub mod cron;
pub mod system_tasks;

3
apps/recorder/src/graphql/domains/subscriber_tasks.rs
View File

@@ -30,8 +30,9 @@ use crate::{
},
},
},
migrations::defs::{ApalisJobs, ApalisSchema},
models::subscriber_tasks,
task::{ApalisJobs, ApalisSchema, SubscriberTaskTrait},
task::SubscriberTaskTrait,
};
fn skip_columns_for_entity_input(context: &mut BuilderContext) {

248
apps/recorder/src/graphql/domains/system_tasks.rs Normal file
View File

@@ -0,0 +1,248 @@
use std::{ops::Deref, sync::Arc};
use async_graphql::dynamic::{FieldValue, Scalar, TypeRef};
use convert_case::Case;
use sea_orm::{
ActiveModelBehavior, ColumnTrait, ConnectionTrait, EntityTrait, Iterable, QueryFilter,
QuerySelect, QueryTrait, prelude::Expr, sea_query::Query,
};
use seaography::{
Builder as SeaographyBuilder, BuilderContext, SeaographyError, prepare_active_model,
};
use ts_rs::TS;
use crate::{
auth::AuthUserInfo,
errors::RecorderError,
graphql::{
domains::subscribers::restrict_subscriber_for_entity,
infra::{
custom::{
generate_entity_create_one_mutation_field,
generate_entity_default_basic_entity_object,
generate_entity_default_insert_input_object, generate_entity_delete_mutation_field,
generate_entity_filtered_mutation_field, register_entity_default_readonly,
},
json::{convert_jsonb_output_for_entity, restrict_jsonb_filter_input_for_entity},
name::{
get_entity_and_column_name, get_entity_basic_type_name,
get_entity_custom_mutation_field_name,
},
},
},
migrations::defs::{ApalisJobs, ApalisSchema},
models::system_tasks,
task::SystemTaskTrait,
};
fn skip_columns_for_entity_input(context: &mut BuilderContext) {
for column in system_tasks::Column::iter() {
if matches!(
column,
system_tasks::Column::Job | system_tasks::Column::SubscriberId
) {
continue;
}
let entity_column_key =
get_entity_and_column_name::<system_tasks::Entity>(context, &column);
context.entity_input.insert_skips.push(entity_column_key);
}
}
pub fn restrict_system_tasks_for_entity<T>(context: &mut BuilderContext, column: &T::Column)
where
T: EntityTrait,
<T as EntityTrait>::Model: Sync,
{
let entity_and_column = get_entity_and_column_name::<T>(context, column);
restrict_jsonb_filter_input_for_entity::<T>(context, column);
convert_jsonb_output_for_entity::<T>(context, column, Some(Case::Camel));
let entity_column_name = get_entity_and_column_name::<T>(context, column);
context.types.input_type_overwrites.insert(
entity_column_name.clone(),
TypeRef::Named(system_tasks::SystemTask::ident().into()),
);
context.types.output_type_overwrites.insert(
entity_column_name.clone(),
TypeRef::Named(system_tasks::SystemTask::ident().into()),
);
context.types.input_conversions.insert(
entity_column_name.clone(),
Box::new(move |resolve_context, value_accessor| {
let task: system_tasks::SystemTaskInput = value_accessor.deserialize()?;
let subscriber_id = resolve_context
.data::<AuthUserInfo>()?
.subscriber_auth
.subscriber_id;
let task = system_tasks::SystemTask::from_input(task, Some(subscriber_id));
let json_value = serde_json::to_value(task).map_err(|err| {
SeaographyError::TypeConversionError(
err.to_string(),
format!("Json - {entity_column_name}"),
)
})?;
Ok(sea_orm::Value::Json(Some(Box::new(json_value))))
}),
);
context.entity_input.update_skips.push(entity_and_column);
}
pub fn register_system_tasks_to_schema_context(context: &mut BuilderContext) {
restrict_subscriber_for_entity::<system_tasks::Entity>(
context,
&system_tasks::Column::SubscriberId,
);
restrict_system_tasks_for_entity::<system_tasks::Entity>(context, &system_tasks::Column::Job);
skip_columns_for_entity_input(context);
}
pub fn register_system_tasks_to_schema_builder(
mut builder: SeaographyBuilder,
) -> SeaographyBuilder {
builder.schema = builder.schema.register(
Scalar::new(system_tasks::SystemTask::ident())
.description(system_tasks::SystemTask::decl()),
);
builder.register_enumeration::<system_tasks::SystemTaskType>();
builder = register_entity_default_readonly!(builder, system_tasks);
let builder_context = builder.context;
{
builder
.outputs
.push(generate_entity_default_basic_entity_object::<
system_tasks::Entity,
>(builder_context));
}
{
let delete_mutation = generate_entity_delete_mutation_field::<system_tasks::Entity>(
builder_context,
Arc::new(|_resolver_ctx, app_ctx, filters| {
Box::pin(async move {
let db = app_ctx.db();
let select_subquery = system_tasks::Entity::find()
.select_only()
.column(system_tasks::Column::Id)
.filter(filters);
let delete_query = Query::delete()
.from_table((ApalisSchema::Schema, ApalisJobs::Table))
.and_where(
Expr::col(ApalisJobs::Id).in_subquery(select_subquery.into_query()),
)
.to_owned();
let db_backend = db.deref().get_database_backend();
let delete_statement = db_backend.build(&delete_query);
let result = db.execute(delete_statement).await?;
Ok::<_, RecorderError>(result.rows_affected())
})
}),
);
builder.mutations.push(delete_mutation);
}
{
let entity_retry_one_mutation_name = get_entity_custom_mutation_field_name::<
system_tasks::Entity,
>(builder_context, "RetryOne");
let retry_one_mutation =
generate_entity_filtered_mutation_field::<system_tasks::Entity, _, _>(
builder_context,
entity_retry_one_mutation_name,
TypeRef::named_nn(get_entity_basic_type_name::<system_tasks::Entity>(
builder_context,
)),
Arc::new(|_resolver_ctx, app_ctx, filters| {
Box::pin(async move {
let db = app_ctx.db();
let job_id = system_tasks::Entity::find()
.filter(filters)
.select_only()
.column(system_tasks::Column::Id)
.into_tuple::<String>()
.one(db)
.await?
.ok_or_else(|| {
RecorderError::from_entity_not_found::<system_tasks::Entity>()
})?;
let task = app_ctx.task();
task.retry_subscriber_task(job_id.clone()).await?;
let task_model = system_tasks::Entity::find()
.filter(system_tasks::Column::Id.eq(&job_id))
.one(db)
.await?
.ok_or_else(|| {
RecorderError::from_entity_not_found::<system_tasks::Entity>()
})?;
Ok::<_, RecorderError>(Some(FieldValue::owned_any(task_model)))
})
}),
);
builder.mutations.push(retry_one_mutation);
}
{
builder
.inputs
.push(generate_entity_default_insert_input_object::<
system_tasks::Entity,
>(builder_context));
let create_one_mutation = generate_entity_create_one_mutation_field::<system_tasks::Entity>(
builder_context,
Arc::new(move |resolver_ctx, app_ctx, input_object| {
Box::pin(async move {
let active_model: Result<system_tasks::ActiveModel, _> =
prepare_active_model(builder_context, &input_object, resolver_ctx);
let task_service = app_ctx.task();
let active_model = active_model?;
let db = app_ctx.db();
let active_model = active_model.before_save(db, true).await?;
let task = active_model.job.unwrap();
let subscriber_id = active_model.subscriber_id.unwrap();
if task.get_subscriber_id() != subscriber_id {
Err(async_graphql::Error::new(
"subscriber_id does not match with job.subscriber_id",
))?;
}
let task_id = task_service.add_system_task(task).await?.to_string();
let db = app_ctx.db();
let task = system_tasks::Entity::find()
.filter(system_tasks::Column::Id.eq(&task_id))
.one(db)
.await?
.ok_or_else(|| {
RecorderError::from_entity_not_found::<system_tasks::Entity>()
})?;
Ok::<_, RecorderError>(task)
})
}),
);
builder.mutations.push(create_one_mutation);
}
builder
}

5
apps/recorder/src/graphql/schema.rs
View File

@@ -39,6 +39,9 @@ use crate::{
subscriptions::{
register_subscriptions_to_schema_builder, register_subscriptions_to_schema_context,
},
system_tasks::{
register_system_tasks_to_schema_builder, register_system_tasks_to_schema_context,
},
},
infra::{
json::register_jsonb_input_filter_to_schema_builder,
@@ -79,6 +82,7 @@ pub fn build_schema(
register_subscription_episode_to_schema_context(&mut context);
register_bangumi_to_schema_context(&mut context);
register_cron_to_schema_context(&mut context);
register_system_tasks_to_schema_context(&mut context);
}
context
});
@@ -103,6 +107,7 @@ pub fn build_schema(
builder = register_subscriber_tasks_to_schema_builder(builder);
builder = register_bangumi_to_schema_builder(builder);
builder = register_cron_to_schema_builder(builder);
builder = register_system_tasks_to_schema_builder(builder);
}
let schema = builder.schema_builder();

4
apps/recorder/src/lib.rs
View File

@@ -27,6 +27,8 @@ pub mod migrations;
pub mod models;
pub mod storage;
pub mod task;
pub mod test_utils;
pub mod utils;
pub mod web;
#[cfg(any(test, feature = "test-utils"))]
pub mod test_utils;

16
apps/recorder/src/media/config.rs
View File

@@ -1,6 +1,8 @@
use serde::{Deserialize, Serialize};
use ts_rs::TS;
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, TS)]
#[ts(rename_all = "camelCase")]
pub enum AutoOptimizeImageFormat {
#[serde(rename = "image/webp")]
Webp,
@@ -10,25 +12,29 @@ pub enum AutoOptimizeImageFormat {
Jxl,
}
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
#[derive(Clone, Debug, Serialize, Deserialize, Default, TS, PartialEq)]
#[ts(rename_all = "camelCase")]
pub struct EncodeWebpOptions {
pub quality: Option<f32>,
}
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
#[derive(Clone, Debug, Serialize, Deserialize, Default, TS, PartialEq)]
#[ts(rename_all = "camelCase")]
pub struct EncodeAvifOptions {
pub quality: Option<u8>,
pub speed: Option<u8>,
pub threads: Option<u8>,
}
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
#[derive(Clone, Debug, Serialize, Deserialize, Default, TS, PartialEq)]
#[ts(rename_all = "camelCase")]
pub struct EncodeJxlOptions {
pub quality: Option<f32>,
pub speed: Option<u8>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, Serialize, Deserialize, TS, PartialEq)]
#[ts(tag = "mimeType")]
#[serde(tag = "mime_type")]
pub enum EncodeImageOptions {
#[serde(rename = "image/webp")]

31
apps/recorder/src/migrations/defs.rs
View File

@@ -190,6 +190,37 @@ pub enum Cron {
Priority,
Status,
SubscriberTask,
SystemTask,
}
#[derive(sea_query::Iden)]
pub enum ApalisSchema {
#[iden = "apalis"]
Schema,
}
#[derive(DeriveIden)]
pub enum ApalisJobs {
#[sea_orm(iden = "jobs")]
Table,
SubscriberId,
SubscriptionId,
Job,
JobType,
Status,
TaskType,
Id,
Attempts,
MaxAttempts,
RunAt,
LastError,
LockAt,
LockBy,
DoneAt,
Priority,
CronId,
}
macro_rules! create_postgres_enum_for_active_enum {

219
apps/recorder/src/migrations/m20250520_021135_add_tasks.rs Normal file
View File

@@ -0,0 +1,219 @@
use async_trait::async_trait;
use sea_orm_migration::{prelude::*, schema::*};
use super::defs::{ApalisJobs, ApalisSchema};
use crate::{
migrations::defs::{Subscribers, Subscriptions},
task::{
SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME,
SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_TRIGGER_NAME, SUBSCRIBER_TASK_APALIS_NAME,
SYSTEM_TASK_APALIS_NAME,
},
};
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.alter_table(
TableAlterStatement::new()
.table((ApalisSchema::Schema, ApalisJobs::Table))
.add_column_if_not_exists(integer_null(ApalisJobs::SubscriberId))
.add_column_if_not_exists(integer_null(ApalisJobs::SubscriptionId))
.add_column_if_not_exists(string_null(ApalisJobs::TaskType))
.add_foreign_key(
TableForeignKey::new()
.name("fk_apalis_jobs_subscriber_id")
.from_tbl((ApalisSchema::Schema, ApalisJobs::Table))
.from_col(ApalisJobs::SubscriberId)
.to_tbl(Subscribers::Table)
.to_col(Subscribers::Id)
.on_delete(ForeignKeyAction::Cascade)
.on_update(ForeignKeyAction::Restrict),
)
.add_foreign_key(
TableForeignKey::new()
.name("fk_apalis_jobs_subscription_id")
.from_tbl((ApalisSchema::Schema, ApalisJobs::Table))
.from_col(ApalisJobs::SubscriptionId)
.to_tbl(Subscriptions::Table)
.to_col(Subscriptions::Id)
.on_delete(ForeignKeyAction::Cascade)
.on_update(ForeignKeyAction::Restrict),
)
.to_owned(),
)
.await?;
let db = manager.get_connection();
db.execute_unprepared(&format!(
r#"UPDATE {apalis_schema}.{apalis_table} SET {subscriber_id} = ({job} ->> '{subscriber_id}')::integer, {task_type} = ({job} ->> '{task_type}')::text, {subscription_id} = ({job} ->> '{subscription_id}')::integer"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
job = ApalisJobs::Job.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
subscription_id = ApalisJobs::SubscriptionId.to_string(),
)).await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE FUNCTION {SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME}() RETURNS trigger AS $$
DECLARE
new_job_subscriber_id integer;
new_job_subscription_id integer;
new_job_task_type text;
BEGIN
new_job_subscriber_id = (NEW.{job} ->> '{subscriber_id}')::integer;
new_job_subscription_id = (NEW.{job} ->> '{subscription_id}')::integer;
new_job_task_type = (NEW.{job} ->> '{task_type}')::text;
IF new_job_subscriber_id != (OLD.{job} ->> '{subscriber_id}')::integer AND new_job_subscriber_id != NEW.{subscriber_id} THEN
NEW.{subscriber_id} = new_job_subscriber_id;
END IF;
IF new_job_subscription_id != (OLD.{job} ->> '{subscription_id}')::integer AND new_job_subscription_id != NEW.{subscription_id} THEN
NEW.{subscription_id} = new_job_subscription_id;
END IF;
IF new_job_task_type != (OLD.{job} ->> '{task_type}')::text AND new_job_task_type != NEW.{task_type} THEN
NEW.{task_type} = new_job_task_type;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;"#,
job = ApalisJobs::Job.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
subscription_id = ApalisJobs::SubscriptionId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
)).await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE TRIGGER {SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_TRIGGER_NAME}
BEFORE INSERT OR UPDATE ON {apalis_schema}.{apalis_table}
FOR EACH ROW
EXECUTE FUNCTION {SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME}();"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string()
))
.await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW subscriber_tasks AS
SELECT
{job},
{job_type},
{status},
{subscriber_id},
{task_type},
{id},
{attempts},
{max_attempts},
{run_at},
{last_error},
{lock_at},
{lock_by},
{done_at},
{priority},
{subscription_id}
FROM {apalis_schema}.{apalis_table}
WHERE {job_type} = '{SUBSCRIBER_TASK_APALIS_NAME}'
AND jsonb_path_exists({job}, '$.{subscriber_id} ? (@.type() == "number")')
AND jsonb_path_exists({job}, '$.{task_type} ? (@.type() == "string")')"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
job = ApalisJobs::Job.to_string(),
job_type = ApalisJobs::JobType.to_string(),
status = ApalisJobs::Status.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
id = ApalisJobs::Id.to_string(),
attempts = ApalisJobs::Attempts.to_string(),
max_attempts = ApalisJobs::MaxAttempts.to_string(),
run_at = ApalisJobs::RunAt.to_string(),
last_error = ApalisJobs::LastError.to_string(),
lock_at = ApalisJobs::LockAt.to_string(),
lock_by = ApalisJobs::LockBy.to_string(),
done_at = ApalisJobs::DoneAt.to_string(),
priority = ApalisJobs::Priority.to_string(),
subscription_id = ApalisJobs::SubscriptionId.to_string(),
))
.await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW system_tasks AS
SELECT
{job},
{job_type},
{status},
{subscriber_id},
{task_type},
{id},
{attempts},
{max_attempts},
{run_at},
{last_error},
{lock_at},
{lock_by},
{done_at},
{priority}
FROM {apalis_schema}.{apalis_table}
WHERE {job_type} = '{SYSTEM_TASK_APALIS_NAME}'
AND jsonb_path_exists({job}, '$.{task_type} ? (@.type() == "string")')"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
job = ApalisJobs::Job.to_string(),
job_type = ApalisJobs::JobType.to_string(),
status = ApalisJobs::Status.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
id = ApalisJobs::Id.to_string(),
attempts = ApalisJobs::Attempts.to_string(),
max_attempts = ApalisJobs::MaxAttempts.to_string(),
run_at = ApalisJobs::RunAt.to_string(),
last_error = ApalisJobs::LastError.to_string(),
lock_at = ApalisJobs::LockAt.to_string(),
lock_by = ApalisJobs::LockBy.to_string(),
done_at = ApalisJobs::DoneAt.to_string(),
priority = ApalisJobs::Priority.to_string(),
))
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
let db = manager.get_connection();
db.execute_unprepared("DROP VIEW IF EXISTS subscriber_tasks")
.await?;
db.execute_unprepared("DROP VIEW IF EXISTS system_tasks")
.await?;
db.execute_unprepared(&format!(
r#"DROP TRIGGER IF EXISTS {SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_TRIGGER_NAME} ON {apalis_schema}.{apalis_table}"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string()
)).await?;
db.execute_unprepared(&format!(
r#"DROP FUNCTION IF EXISTS {SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME}()"#,
))
.await?;
manager
.alter_table(
TableAlterStatement::new()
.table((ApalisSchema::Schema, ApalisJobs::Table))
.drop_foreign_key("fk_apalis_jobs_subscriber_id")
.drop_foreign_key("fk_apalis_jobs_subscription_id")
.drop_column(ApalisJobs::SubscriberId)
.drop_column(ApalisJobs::SubscriptionId)
.to_owned(),
)
.await?;
Ok(())
}
}

64
apps/recorder/src/migrations/m20250520_021135_subscriber_tasks.rs
View File

@@ -1,64 +0,0 @@
use async_trait::async_trait;
use sea_orm_migration::prelude::*;
use crate::task::SUBSCRIBER_TASK_APALIS_NAME;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
let db = manager.get_connection();
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW subscriber_tasks AS
SELECT
job,
job_type,
status,
(job ->> 'subscriber_id'::text)::integer AS subscriber_id,
job ->> 'task_type'::text AS task_type,
id,
attempts,
max_attempts,
run_at,
last_error,
lock_at,
lock_by,
done_at,
priority
FROM apalis.jobs
WHERE job_type = '{SUBSCRIBER_TASK_APALIS_NAME}'
AND jsonb_path_exists(job, '$.subscriber_id ? (@.type() == "number")')
AND jsonb_path_exists(job, '$.task_type ? (@.type() == "string")')"#,
))
.await?;
db.execute_unprepared(&format!(
r#"CREATE INDEX IF NOT EXISTS idx_apalis_jobs_subscriber_id
ON apalis.jobs (((job -> 'subscriber_id')::integer))
WHERE job_type = '{SUBSCRIBER_TASK_APALIS_NAME}'
AND jsonb_path_exists(job, '$.subscriber_id ? (@.type() == "number")')
AND jsonb_path_exists(job, '$.task_type ? (@.type() == "string")')"#
))
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
let db = manager.get_connection();
db.execute_unprepared(
r#"DROP INDEX IF EXISTS idx_apalis_jobs_subscriber_id
ON apalis.jobs"#,
)
.await?;
db.execute_unprepared("DROP VIEW IF EXISTS subscriber_tasks")
.await?;
Ok(())
}
}

62
apps/recorder/src/migrations/m20250625_060701_add_subscription_id_to_subscriber_tasks.rs
View File

@@ -1,62 +0,0 @@
use async_trait::async_trait;
use sea_orm_migration::prelude::*;
use crate::task::SUBSCRIBER_TASK_APALIS_NAME;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
let db = manager.get_connection();
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW subscriber_tasks AS
SELECT
job,
job_type,
status,
(job ->> 'subscriber_id')::integer AS subscriber_id,
job ->> 'task_type' AS task_type,
id,
attempts,
max_attempts,
run_at,
last_error,
lock_at,
lock_by,
done_at,
priority,
(job ->> 'subscription_id')::integer AS subscription_id
FROM apalis.jobs
WHERE job_type = '{SUBSCRIBER_TASK_APALIS_NAME}'
AND jsonb_path_exists(job, '$.subscriber_id ? (@.type() == "number")')
AND jsonb_path_exists(job, '$.task_type ? (@.type() == "string")')"#,
))
.await?;
db.execute_unprepared(&format!(
r#"CREATE INDEX IF NOT EXISTS idx_apalis_jobs_subscription_id
ON apalis.jobs (((job -> 'subscription_id')::integer))
WHERE job_type = '{SUBSCRIBER_TASK_APALIS_NAME}'
AND jsonb_path_exists(job, '$.subscription_id ? (@.type() == "number")')
AND jsonb_path_exists(job, '$.task_type ? (@.type() == "string")')"#
))
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
let db = manager.get_connection();
db.execute_unprepared(
r#"DROP INDEX IF EXISTS idx_apalis_jobs_subscription_id
ON apalis.jobs"#,
)
.await?;
Ok(())
}
}

295
apps/recorder/src/migrations/m20250629_065628_add_cron.rs
View File

@@ -4,13 +4,18 @@ use sea_orm_migration::{prelude::*, schema::*};
use crate::{
migrations::defs::{
Cron, CustomSchemaManagerExt, GeneralIds, Subscribers, Subscriptions, table_auto_z,
ApalisJobs, ApalisSchema, Cron, CustomSchemaManagerExt, GeneralIds, Subscribers,
Subscriptions, table_auto_z,
},
models::cron::{
CHECK_AND_TRIGGER_DUE_CRONS_FUNCTION_NAME, CRON_DUE_EVENT, CronStatus, CronStatusEnum,
NOTIFY_DUE_CRON_WHEN_MUTATING_FUNCTION_NAME, NOTIFY_DUE_CRON_WHEN_MUTATING_TRIGGER_NAME,
SETUP_CRON_EXTRA_FOREIGN_KEYS_FUNCTION_NAME, SETUP_CRON_EXTRA_FOREIGN_KEYS_TRIGGER_NAME,
},
task::{
SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME, SUBSCRIBER_TASK_APALIS_NAME,
SYSTEM_TASK_APALIS_NAME,
},
};
#[derive(DeriveMigrationName)]
@@ -52,6 +57,7 @@ impl MigrationTrait for Migration {
CronStatus::iden_values(),
))
.col(json_binary_null(Cron::SubscriberTask))
.col(json_binary_null(Cron::SystemTask))
.foreign_key(
ForeignKey::create()
.name("fk_cron_subscriber_id")
@@ -91,12 +97,22 @@ impl MigrationTrait for Migration {
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE FUNCTION {SETUP_CRON_EXTRA_FOREIGN_KEYS_FUNCTION_NAME}() RETURNS trigger AS $$
DECLARE
new_subscriber_task_subscriber_id integer;
new_subscriber_task_subscription_id integer;
new_system_task_subscriber_id integer;
BEGIN
IF jsonb_path_exists(NEW.{subscriber_task}, '$.subscriber_id ? (@.type() == "number")') THEN
NEW.{subscriber_id} = (NEW.{subscriber_task} ->> 'subscriber_id')::integer;
new_subscriber_task_subscriber_id = (NEW.{subscriber_task} ->> 'subscriber_id')::integer;
new_subscriber_task_subscription_id = (NEW.{subscriber_task} ->> 'subscription_id')::integer;
new_system_task_subscriber_id = (NEW.{system_task} ->> 'subscriber_id')::integer;
IF new_subscriber_task_subscriber_id != (OLD.{subscriber_task} ->> 'subscriber_id')::integer AND new_subscriber_task_subscriber_id != NEW.{subscriber_id} THEN
NEW.{subscriber_id} = new_subscriber_task_subscriber_id;
END IF;
IF jsonb_path_exists(NEW.{subscriber_task}, '$.subscription_id ? (@.type() == "number")') THEN
NEW.{subscription_id} = (NEW.{subscriber_task} ->> 'subscription_id')::integer;
IF new_subscriber_task_subscription_id != (OLD.{subscriber_task} ->> 'subscription_id')::integer AND new_subscriber_task_subscription_id != NEW.{subscription_id} THEN
NEW.{subscription_id} = new_subscriber_task_subscription_id;
END IF;
IF new_system_task_subscriber_id != (OLD.{system_task} ->> 'subscriber_id')::integer AND new_system_task_subscriber_id != NEW.{subscriber_id} THEN
NEW.{subscriber_id} = new_system_task_subscriber_id;
END IF;
RETURN NEW;
END;
@@ -104,6 +120,7 @@ impl MigrationTrait for Migration {
subscriber_task = &Cron::SubscriberTask.to_string(),
subscriber_id = &Cron::SubscriberId.to_string(),
subscription_id = &Cron::SubscriptionId.to_string(),
system_task = &Cron::SystemTask.to_string(),
)).await?;
db.execute_unprepared(&format!(
@@ -208,12 +225,280 @@ impl MigrationTrait for Migration {
))
.await?;
manager
.alter_table(
TableAlterStatement::new()
.table((ApalisSchema::Schema, ApalisJobs::Table))
.add_column_if_not_exists(integer_null(ApalisJobs::CronId))
.add_foreign_key(
TableForeignKey::new()
.name("fk_apalis_jobs_cron_id")
.from_tbl((ApalisSchema::Schema, ApalisJobs::Table))
.from_col(ApalisJobs::CronId)
.to_tbl(Cron::Table)
.to_col(Cron::Id)
.on_delete(ForeignKeyAction::NoAction)
.on_update(ForeignKeyAction::NoAction),
)
.to_owned(),
)
.await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW subscriber_tasks AS
SELECT
{job},
{job_type},
{status},
{subscriber_id},
{task_type},
{id},
{attempts},
{max_attempts},
{run_at},
{last_error},
{lock_at},
{lock_by},
{done_at},
{priority},
{subscription_id},
{cron_id}
FROM {apalis_schema}.{apalis_table}
WHERE {job_type} = '{SUBSCRIBER_TASK_APALIS_NAME}'
AND jsonb_path_exists({job}, '$.{subscriber_id} ? (@.type() == "number")')
AND jsonb_path_exists({job}, '$.{task_type} ? (@.type() == "string")')"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
job = ApalisJobs::Job.to_string(),
job_type = ApalisJobs::JobType.to_string(),
status = ApalisJobs::Status.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
id = ApalisJobs::Id.to_string(),
attempts = ApalisJobs::Attempts.to_string(),
max_attempts = ApalisJobs::MaxAttempts.to_string(),
run_at = ApalisJobs::RunAt.to_string(),
last_error = ApalisJobs::LastError.to_string(),
lock_at = ApalisJobs::LockAt.to_string(),
lock_by = ApalisJobs::LockBy.to_string(),
done_at = ApalisJobs::DoneAt.to_string(),
priority = ApalisJobs::Priority.to_string(),
subscription_id = ApalisJobs::SubscriptionId.to_string(),
cron_id = ApalisJobs::CronId.to_string(),
))
.await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW system_tasks AS
SELECT
{job},
{job_type},
{status},
{subscriber_id},
{task_type},
{id},
{attempts},
{max_attempts},
{run_at},
{last_error},
{lock_at},
{lock_by},
{done_at},
{priority},
{cron_id}
FROM {apalis_schema}.{apalis_table}
WHERE {job_type} = '{SYSTEM_TASK_APALIS_NAME}'
AND jsonb_path_exists({job}, '$.{task_type} ? (@.type() == "string")')"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
job = ApalisJobs::Job.to_string(),
job_type = ApalisJobs::JobType.to_string(),
status = ApalisJobs::Status.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
id = ApalisJobs::Id.to_string(),
attempts = ApalisJobs::Attempts.to_string(),
max_attempts = ApalisJobs::MaxAttempts.to_string(),
run_at = ApalisJobs::RunAt.to_string(),
last_error = ApalisJobs::LastError.to_string(),
lock_at = ApalisJobs::LockAt.to_string(),
lock_by = ApalisJobs::LockBy.to_string(),
done_at = ApalisJobs::DoneAt.to_string(),
priority = ApalisJobs::Priority.to_string(),
cron_id = ApalisJobs::CronId.to_string(),
))
.await?;
db.execute_unprepared(&format!(
r#"
UPDATE {apalis_schema}.{apalis_table} SET {cron_id} = ({job} ->> '{cron_id}')::integer
"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
job = ApalisJobs::Job.to_string(),
cron_id = ApalisJobs::CronId.to_string(),
))
.await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE FUNCTION {SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME}() RETURNS trigger AS $$
DECLARE
new_job_subscriber_id integer;
new_job_subscription_id integer;
new_job_cron_id integer;
new_job_task_type text;
BEGIN
new_job_subscriber_id = (NEW.{job} ->> '{subscriber_id}')::integer;
new_job_subscription_id = (NEW.{job} ->> '{subscription_id}')::integer;
new_job_cron_id = (NEW.{job} ->> '{cron_id}')::integer;
new_job_task_type = (NEW.{job} ->> '{task_type}')::text;
IF new_job_subscriber_id != (OLD.{job} ->> '{subscriber_id}')::integer AND new_job_subscriber_id != NEW.{subscriber_id} THEN
NEW.{subscriber_id} = new_job_subscriber_id;
END IF;
IF new_job_subscription_id != (OLD.{job} ->> '{subscription_id}')::integer AND new_job_subscription_id != NEW.{subscription_id} THEN
NEW.{subscription_id} = new_job_subscription_id;
END IF;
IF new_job_cron_id != (OLD.{job} ->> '{cron_id}')::integer AND new_job_cron_id != NEW.{cron_id} THEN
NEW.{cron_id} = new_job_cron_id;
END IF;
IF new_job_task_type != (OLD.{job} ->> '{task_type}')::text AND new_job_task_type != NEW.{task_type} THEN
NEW.{task_type} = new_job_task_type;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;"#,
job = ApalisJobs::Job.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
subscription_id = ApalisJobs::SubscriptionId.to_string(),
cron_id = ApalisJobs::CronId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
)).await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
let db = manager.get_connection();
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE FUNCTION {SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME}() RETURNS trigger AS $$
DECLARE
new_job_subscriber_id integer;
new_job_subscription_id integer;
new_job_task_type text;
BEGIN
new_job_subscriber_id = (NEW.{job} ->> '{subscriber_id}')::integer;
new_job_subscription_id = (NEW.{job} ->> '{subscription_id}')::integer;
new_job_task_type = (NEW.{job} ->> '{task_type}')::text;
IF new_job_subscriber_id != (OLD.{job} ->> '{subscriber_id}')::integer AND new_job_subscriber_id != NEW.{subscriber_id} THEN
NEW.{subscriber_id} = new_job_subscriber_id;
END IF;
IF new_job_subscription_id != (OLD.{job} ->> '{subscription_id}')::integer AND new_job_subscription_id != NEW.{subscription_id} THEN
NEW.{subscription_id} = new_job_subscription_id;
END IF;
IF new_job_task_type != (OLD.{job} ->> '{task_type}')::text AND new_job_task_type != NEW.{task_type} THEN
NEW.{task_type} = new_job_task_type;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;"#,
job = ApalisJobs::Job.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
subscription_id = ApalisJobs::SubscriptionId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
)).await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW subscriber_tasks AS
SELECT
{job},
{job_type},
{status},
{subscriber_id},
{task_type},
{id},
{attempts},
{max_attempts},
{run_at},
{last_error},
{lock_at},
{lock_by},
{done_at},
{priority},
{subscription_id}
FROM {apalis_schema}.{apalis_table}
WHERE {job_type} = '{SUBSCRIBER_TASK_APALIS_NAME}'
AND jsonb_path_exists({job}, '$.{subscriber_id} ? (@.type() == "number")')
AND jsonb_path_exists({job}, '$.{task_type} ? (@.type() == "string")')"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
job = ApalisJobs::Job.to_string(),
job_type = ApalisJobs::JobType.to_string(),
status = ApalisJobs::Status.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
id = ApalisJobs::Id.to_string(),
attempts = ApalisJobs::Attempts.to_string(),
max_attempts = ApalisJobs::MaxAttempts.to_string(),
run_at = ApalisJobs::RunAt.to_string(),
last_error = ApalisJobs::LastError.to_string(),
lock_at = ApalisJobs::LockAt.to_string(),
lock_by = ApalisJobs::LockBy.to_string(),
done_at = ApalisJobs::DoneAt.to_string(),
priority = ApalisJobs::Priority.to_string(),
subscription_id = ApalisJobs::SubscriptionId.to_string(),
))
.await?;
db.execute_unprepared(&format!(
r#"CREATE OR REPLACE VIEW system_tasks AS
SELECT
{job},
{job_type},
{status},
{subscriber_id},
{task_type},
{id},
{attempts},
{max_attempts},
{run_at},
{last_error},
{lock_at},
{lock_by},
{done_at},
{priority}
FROM {apalis_schema}.{apalis_table}
WHERE {job_type} = '{SYSTEM_TASK_APALIS_NAME}'
AND jsonb_path_exists({job}, '$.{task_type} ? (@.type() == "string")')"#,
apalis_schema = ApalisSchema::Schema.to_string(),
apalis_table = ApalisJobs::Table.to_string(),
job = ApalisJobs::Job.to_string(),
job_type = ApalisJobs::JobType.to_string(),
status = ApalisJobs::Status.to_string(),
subscriber_id = ApalisJobs::SubscriberId.to_string(),
task_type = ApalisJobs::TaskType.to_string(),
id = ApalisJobs::Id.to_string(),
attempts = ApalisJobs::Attempts.to_string(),
max_attempts = ApalisJobs::MaxAttempts.to_string(),
run_at = ApalisJobs::RunAt.to_string(),
last_error = ApalisJobs::LastError.to_string(),
lock_at = ApalisJobs::LockAt.to_string(),
lock_by = ApalisJobs::LockBy.to_string(),
done_at = ApalisJobs::DoneAt.to_string(),
priority = ApalisJobs::Priority.to_string(),
))
.await?;
manager
.alter_table(
TableAlterStatement::new()
.table((ApalisSchema::Schema, ApalisJobs::Table))
.drop_column(ApalisJobs::CronId)
.drop_foreign_key("fk_apalis_jobs_cron_id")
.to_owned(),
)
.await?;
db.execute_unprepared(&format!(
r#"DROP TRIGGER IF EXISTS {NOTIFY_DUE_CRON_WHEN_MUTATING_TRIGGER_NAME} ON {table};"#,
table = &Cron::Table.to_string(),

6
apps/recorder/src/migrations/mod.rs
View File

@@ -7,10 +7,9 @@ pub mod m20220101_000001_init;
pub mod m20240224_082543_add_downloads;
pub mod m20241231_000001_auth;
pub mod m20250501_021523_credential_3rd;
pub mod m20250520_021135_subscriber_tasks;
pub mod m20250520_021135_add_tasks;
pub mod m20250622_015618_feeds;
pub mod m20250622_020819_bangumi_and_episode_type;
pub mod m20250625_060701_add_subscription_id_to_subscriber_tasks;
pub mod m20250629_065628_add_cron;
pub struct Migrator;
@@ -23,10 +22,9 @@ impl MigratorTrait for Migrator {
Box::new(m20240224_082543_add_downloads::Migration),
Box::new(m20241231_000001_auth::Migration),
Box::new(m20250501_021523_credential_3rd::Migration),
Box::new(m20250520_021135_subscriber_tasks::Migration),
Box::new(m20250520_021135_add_tasks::Migration),
Box::new(m20250622_015618_feeds::Migration),
Box::new(m20250622_020819_bangumi_and_episode_type::Migration),
Box::new(m20250625_060701_add_subscription_id_to_subscriber_tasks::Migration),
Box::new(m20250629_065628_add_cron::Migration),
]
}

52
apps/recorder/src/models/cron/mod.rs
View File

@@ -21,8 +21,10 @@ use sea_orm::{
use serde::{Deserialize, Serialize};
use crate::{
app::AppContextTrait, errors::RecorderResult, models::subscriber_tasks,
task::SubscriberTaskTrait,
app::AppContextTrait,
errors::RecorderResult,
models::{subscriber_tasks, system_tasks},
task::{SubscriberTaskTrait, SystemTaskTrait},
};
#[derive(
@@ -41,7 +43,7 @@ pub enum CronStatus {
Failed,
}
#[derive(Debug, Clone, PartialEq, Eq, DeriveEntityModel, Serialize, Deserialize)]
#[derive(Debug, Clone, DeriveEntityModel, PartialEq, Serialize, Deserialize)]
#[sea_orm(table_name = "cron")]
pub struct Model {
#[sea_orm(default_expr = "Expr::current_timestamp()")]
@@ -70,6 +72,7 @@ pub struct Model {
#[sea_orm(default_expr = "true")]
pub enabled: bool,
pub subscriber_task: Option<subscriber_tasks::SubscriberTask>,
pub system_task: Option<system_tasks::SystemTask>,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
@@ -79,7 +82,7 @@ pub enum Relation {
from = "Column::SubscriberId",
to = "super::subscribers::Column::Id",
on_update = "Cascade",
on_delete = "Cascade"
on_delete = "Restrict"
)]
Subscriber,
#[sea_orm(
@@ -87,9 +90,13 @@ pub enum Relation {
from = "Column::SubscriptionId",
to = "super::subscriptions::Column::Id",
on_update = "Cascade",
on_delete = "Cascade"
on_delete = "Restrict"
)]
Subscription,
#[sea_orm(has_many = "super::subscriber_tasks::Entity")]
SubscriberTask,
#[sea_orm(has_many = "super::system_tasks::Entity")]
SystemTask,
}
impl Related<super::subscribers::Entity> for Entity {
@@ -104,12 +111,28 @@ impl Related<super::subscriptions::Entity> for Entity {
}
}
impl Related<super::subscriber_tasks::Entity> for Entity {
fn to() -> RelationDef {
Relation::SubscriberTask.def()
}
}
impl Related<super::system_tasks::Entity> for Entity {
fn to() -> RelationDef {
Relation::SystemTask.def()
}
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelatedEntity)]
pub enum RelatedEntity {
#[sea_orm(entity = "super::subscribers::Entity")]
Subscriber,
#[sea_orm(entity = "super::subscriptions::Entity")]
Subscription,
#[sea_orm(entity = "super::subscriber_tasks::Entity")]
SubscriberTask,
#[sea_orm(entity = "super::system_tasks::Entity")]
SystemTask,
}
#[async_trait]
@@ -136,6 +159,14 @@ impl ActiveModelBehavior for ActiveModel {
"Cron subscriber_id does not match subscriber_task.subscriber_id".to_string(),
));
}
if let ActiveValue::Set(Some(subscriber_id)) = self.subscriber_id
&& let ActiveValue::Set(Some(ref system_task)) = self.system_task
&& system_task.get_subscriber_id() != Some(subscriber_id)
{
return Err(DbErr::Custom(
"Cron subscriber_id does not match system_task.subscriber_id".to_string(),
));
}
Ok(self)
}
@@ -219,11 +250,18 @@ impl Model {
async fn exec_cron(&self, ctx: &dyn AppContextTrait) -> RecorderResult<()> {
if let Some(subscriber_task) = self.subscriber_task.as_ref() {
let task_service = ctx.task();
let mut new_subscriber_task = subscriber_task.clone();
new_subscriber_task.set_cron_id(Some(self.id));
task_service
.add_subscriber_task(subscriber_task.clone())
.add_subscriber_task(new_subscriber_task)
.await?;
} else if let Some(system_task) = self.system_task.as_ref() {
let task_service = ctx.task();
let mut new_system_task = system_task.clone();
new_system_task.set_cron_id(Some(self.id));
task_service.add_system_task(new_system_task).await?;
} else {
unimplemented!("Cron without subscriber task is not supported now");
unimplemented!("Cron without unknown task is not supported now");
}
Ok(())

3
apps/recorder/src/models/mod.rs
View File

@@ -1,6 +1,7 @@
pub mod auth;
pub mod bangumi;
pub mod credential_3rd;
pub mod cron;
pub mod downloaders;
pub mod downloads;
pub mod episodes;
@@ -11,4 +12,4 @@ pub mod subscribers;
pub mod subscription_bangumi;
pub mod subscription_episode;
pub mod subscriptions;
pub mod cron;
pub mod system_tasks;

21
apps/recorder/src/models/subscriber_tasks/mod.rs
View File

@@ -24,13 +24,14 @@ pub enum SubscriberTaskStatus {
Killed,
}
#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
#[sea_orm(table_name = "subscriber_tasks")]
pub struct Model {
#[sea_orm(primary_key)]
pub id: String,
pub subscriber_id: i32,
pub subscription_id: Option<i32>,
pub cron_id: Option<i32>,
pub job: SubscriberTask,
pub task_type: SubscriberTaskType,
pub status: SubscriberTaskStatus,
@@ -51,7 +52,7 @@ pub enum Relation {
from = "Column::SubscriberId",
to = "super::subscribers::Column::Id",
on_update = "Cascade",
on_delete = "Cascade"
on_delete = "NoAction"
)]
Subscriber,
#[sea_orm(
@@ -62,6 +63,14 @@ pub enum Relation {
on_delete = "NoAction"
)]
Subscription,
#[sea_orm(
belongs_to = "super::cron::Entity",
from = "Column::CronId",
to = "super::cron::Column::Id",
on_update = "NoAction",
on_delete = "NoAction"
)]
Cron,
}
impl Related<super::subscribers::Entity> for Entity {
@@ -76,12 +85,20 @@ impl Related<super::subscriptions::Entity> for Entity {
}
}
impl Related<super::cron::Entity> for Entity {
fn to() -> RelationDef {
Relation::Cron.def()
}
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelatedEntity)]
pub enum RelatedEntity {
#[sea_orm(entity = "super::subscribers::Entity")]
Subscriber,
#[sea_orm(entity = "super::subscriptions::Entity")]
Subscription,
#[sea_orm(entity = "super::cron::Entity")]
Cron,
}
#[async_trait]

10
apps/recorder/src/models/subscribers.rs
View File

@@ -45,6 +45,8 @@ pub enum Relation {
Feed,
#[sea_orm(has_many = "super::subscriber_tasks::Entity")]
SubscriberTask,
#[sea_orm(has_many = "super::system_tasks::Entity")]
SystemTask,
}
impl Related<super::subscriptions::Entity> for Entity {
@@ -95,6 +97,12 @@ impl Related<super::subscriber_tasks::Entity> for Entity {
}
}
impl Related<super::system_tasks::Entity> for Entity {
fn to() -> RelationDef {
Relation::SystemTask.def()
}
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelatedEntity)]
pub enum RelatedEntity {
#[sea_orm(entity = "super::subscriptions::Entity")]
@@ -111,6 +119,8 @@ pub enum RelatedEntity {
Feed,
#[sea_orm(entity = "super::subscriber_tasks::Entity")]
SubscriberTask,
#[sea_orm(entity = "super::system_tasks::Entity")]
SystemTask,
}
#[derive(Debug, Deserialize, Serialize)]

99
apps/recorder/src/models/system_tasks/mod.rs Normal file
View File

@@ -0,0 +1,99 @@
use async_trait::async_trait;
use sea_orm::{ActiveValue, entity::prelude::*};
pub use crate::task::{
SystemTask, SystemTaskInput, SystemTaskType, SystemTaskTypeEnum, SystemTaskTypeVariant,
SystemTaskTypeVariantIter,
};
#[derive(Clone, Debug, PartialEq, Eq, DeriveActiveEnum, EnumIter, DeriveDisplay)]
#[sea_orm(rs_type = "String", db_type = "Text")]
pub enum SystemTaskStatus {
#[sea_orm(string_value = "Pending")]
Pending,
#[sea_orm(string_value = "Scheduled")]
Scheduled,
#[sea_orm(string_value = "Running")]
Running,
#[sea_orm(string_value = "Done")]
Done,
#[sea_orm(string_value = "Failed")]
Failed,
#[sea_orm(string_value = "Killed")]
Killed,
}
#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
#[sea_orm(table_name = "system_tasks")]
pub struct Model {
#[sea_orm(primary_key)]
pub id: String,
pub subscriber_id: Option<i32>,
pub cron_id: Option<i32>,
pub job: SystemTask,
pub task_type: SystemTaskType,
pub status: SystemTaskStatus,
pub attempts: i32,
pub max_attempts: i32,
pub run_at: DateTimeUtc,
pub last_error: Option<String>,
pub lock_at: Option<DateTimeUtc>,
pub lock_by: Option<String>,
pub done_at: Option<DateTimeUtc>,
pub priority: i32,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {
#[sea_orm(
belongs_to = "super::subscribers::Entity",
from = "Column::SubscriberId",
to = "super::subscribers::Column::Id",
on_update = "Cascade",
on_delete = "Restrict"
)]
Subscriber,
#[sea_orm(
belongs_to = "super::cron::Entity",
from = "Column::CronId",
to = "super::cron::Column::Id",
on_update = "NoAction",
on_delete = "NoAction"
)]
Cron,
}
impl Related<super::subscribers::Entity> for Entity {
fn to() -> RelationDef {
Relation::Subscriber.def()
}
}
impl Related<super::cron::Entity> for Entity {
fn to() -> RelationDef {
Relation::Cron.def()
}
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelatedEntity)]
pub enum RelatedEntity {
#[sea_orm(entity = "super::subscribers::Entity")]
Subscriber,
#[sea_orm(entity = "super::cron::Entity")]
Cron,
}
#[async_trait]
impl ActiveModelBehavior for ActiveModel {
async fn before_save<C>(mut self, _db: &C, _insert: bool) -> Result<Self, DbErr>
where
C: ConnectionTrait,
{
if let ActiveValue::Set(Some(..)) = self.subscriber_id {
return Err(DbErr::Custom(
"SystemTask can not be created by subscribers now".to_string(),
));
}
Ok(self)
}
}

7
apps/recorder/src/storage/client.rs
View File

@@ -89,6 +89,13 @@ impl StorageService {
p
}
#[cfg(any(test, feature = "test-utils"))]
pub fn build_test_path(&self, path: impl AsRef<Path>) -> PathBuf {
let mut p = PathBuf::from("/test");
p.push(path);
p
}
pub fn build_public_path(&self, path: impl AsRef<Path>) -> PathBuf {
let mut p = PathBuf::from("/public");
p.push(path);

32
apps/recorder/src/task/core.rs
View File

@@ -2,12 +2,16 @@ use std::sync::Arc;
use async_trait::async_trait;
use futures::{Stream, StreamExt, pin_mut};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde::{Serialize, de::DeserializeOwned};
use crate::{app::AppContextTrait, errors::RecorderResult};
pub const SYSTEM_TASK_APALIS_NAME: &str = "system_task";
pub const SUBSCRIBER_TASK_APALIS_NAME: &str = "subscriber_task";
pub const SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME: &str =
"setup_apalis_jobs_extra_foreign_keys";
pub const SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_TRIGGER_NAME: &str =
"setup_apalis_jobs_extra_foreign_keys_trigger";
#[async_trait]
pub trait AsyncTaskTrait: Serialize + DeserializeOwned + Sized {
@@ -41,20 +45,30 @@ where
}
}
pub trait SystemTaskTrait: AsyncTaskTrait {
type InputType: Serialize + DeserializeOwned + Sized + Send;
fn get_subscriber_id(&self) -> Option<i32>;
fn set_subscriber_id(&mut self, subscriber_id: Option<i32>);
fn get_cron_id(&self) -> Option<i32>;
fn set_cron_id(&mut self, cron_id: Option<i32>);
fn from_input(input: Self::InputType, subscriber_id: Option<i32>) -> Self;
}
pub trait SubscriberTaskTrait: AsyncTaskTrait {
type InputType: Serialize + DeserializeOwned + Sized + Send;
fn get_subscriber_id(&self) -> i32;
fn set_subscriber_id(&mut self, subscriber_id: i32);
fn get_cron_id(&self) -> Option<i32>;
fn set_cron_id(&mut self, cron_id: Option<i32>);
fn from_input(input: Self::InputType, subscriber_id: i32) -> Self;
}
pub trait SystemTaskTrait: AsyncTaskTrait {}
#[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Clone)]
pub struct SubscriberTaskBase {
pub subscriber_id: i32,
pub cron_id: Option<i32>,
}

16
apps/recorder/src/task/extern.rs
View File

@@ -1,16 +0,0 @@
use sea_orm::sea_query;
#[derive(sea_query::Iden)]
pub enum ApalisSchema {
#[iden = "apalis"]
Schema,
}
#[derive(sea_query::Iden)]
pub enum ApalisJobs {
#[iden = "jobs"]
Table,
Id,
}

13
apps/recorder/src/task/mod.rs
View File

@@ -1,21 +1,22 @@
mod config;
mod core;
mod r#extern;
mod registry;
mod service;
pub use core::{
AsyncTaskTrait, SUBSCRIBER_TASK_APALIS_NAME, SYSTEM_TASK_APALIS_NAME, StreamTaskTrait,
SubscriberTaskBase, SubscriberTaskTrait, SystemTaskTrait,
AsyncTaskTrait, SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_FUNCTION_NAME,
SETUP_APALIS_JOBS_EXTRA_FOREIGN_KEYS_TRIGGER_NAME, SUBSCRIBER_TASK_APALIS_NAME,
SYSTEM_TASK_APALIS_NAME, StreamTaskTrait, SubscriberTaskTrait, SystemTaskTrait,
};
pub use config::TaskConfig;
pub use r#extern::{ApalisJobs, ApalisSchema};
pub use registry::{
OptimizeImageTask, SubscriberTask, SubscriberTaskInput, SubscriberTaskType,
SubscriberTaskTypeEnum, SubscriberTaskTypeVariant, SubscriberTaskTypeVariantIter,
SyncOneSubscriptionFeedsFullTask, SyncOneSubscriptionFeedsIncrementalTask,
SyncOneSubscriptionSourcesTask, SystemTask, SystemTaskType, SystemTaskTypeEnum,
SystemTaskTypeVariant, SystemTaskTypeVariantIter,
SyncOneSubscriptionSourcesTask, SystemTask, SystemTaskInput, SystemTaskType,
SystemTaskTypeEnum, SystemTaskTypeVariant, SystemTaskTypeVariantIter,
};
#[allow(unused_imports)]
pub(crate) use registry::{register_subscriber_task_type, register_system_task_type};
pub use service::TaskService;

6
apps/recorder/src/task/registry/mod.rs
View File

@@ -1,12 +1,14 @@
mod subscriber;
mod system;
pub(crate) use subscriber::register_subscriber_task_type;
pub use subscriber::{
SubscriberTask, SubscriberTaskInput, SubscriberTaskType, SubscriberTaskTypeEnum,
SubscriberTaskTypeVariant, SubscriberTaskTypeVariantIter, SyncOneSubscriptionFeedsFullTask,
SyncOneSubscriptionFeedsIncrementalTask, SyncOneSubscriptionSourcesTask,
};
pub(crate) use system::register_system_task_type;
pub use system::{
OptimizeImageTask, SystemTask, SystemTaskType, SystemTaskTypeEnum, SystemTaskTypeVariant,
SystemTaskTypeVariantIter,
OptimizeImageTask, SystemTask, SystemTaskInput, SystemTaskType, SystemTaskTypeEnum,
SystemTaskTypeVariant, SystemTaskTypeVariantIter,
};

12
apps/recorder/src/task/registry/subscriber/base.rs
View File

@@ -7,7 +7,7 @@ macro_rules! register_subscriber_task_type {
) => {
$(#[$type_meta])*
#[derive(typed_builder::TypedBuilder, ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[ts(export, rename_all = "camelCase")]
#[ts(rename_all = "camelCase")]
$task_vis struct $task_name {
$($(#[$field_meta])* pub $field_name: $field_type,)*
pub subscriber_id: i32,
@@ -20,7 +20,7 @@ macro_rules! register_subscriber_task_type {
$(#[$type_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
#[ts(export, rename_all = "camelCase")]
#[ts(rename_all = "camelCase")]
$task_vis struct [<$task_name Input>] {
$($(#[$field_meta])* pub $field_name: $field_type,)*
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -44,6 +44,14 @@ macro_rules! register_subscriber_task_type {
self.cron_id
}
fn set_subscriber_id(&mut self, subscriber_id: i32) {
self.subscriber_id = subscriber_id;
}
fn set_cron_id(&mut self, cron_id: Option<i32>) {
self.cron_id = cron_id;
}
fn from_input(input: Self::InputType, subscriber_id: i32) -> Self {
Self {
$($field_name: input.$field_name,)*

36
apps/recorder/src/task/registry/subscriber/mod.rs
View File

@@ -1,6 +1,7 @@
mod base;
mod subscription;
pub(crate) use base::register_subscriber_task_type;
use sea_orm::{DeriveActiveEnum, DeriveDisplay, EnumIter, FromJsonQueryResult};
pub use subscription::{
SyncOneSubscriptionFeedsFullTask, SyncOneSubscriptionFeedsIncrementalTask,
@@ -44,7 +45,7 @@ macro_rules! register_subscriber_task_types {
$(#[$task_enum_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[serde(tag = "task_type")]
#[ts(export,rename = "SubscriberTaskType", rename_all = "camelCase", tag = "taskType")]
#[ts(export, rename = "SubscriberTaskType", rename_all = "camelCase", tag = "taskType")]
$task_vis enum $task_enum_name {
$(
$(#[$task_variant_meta])*
@@ -57,7 +58,7 @@ macro_rules! register_subscriber_task_types {
$(#[$task_enum_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[serde(tag = "taskType", rename_all = "camelCase")]
#[ts(export,rename_all = "camelCase", tag = "taskType")]
#[ts(export, rename_all = "camelCase", tag = "taskType")]
$task_vis enum [<$task_enum_name Input>] {
$(
$(#[$task_variant_meta])*
@@ -67,23 +68,6 @@ macro_rules! register_subscriber_task_types {
}
}
impl TryFrom<$task_enum_name> for serde_json::Value {
type Error = $crate::errors::RecorderError;
fn try_from(value: $task_enum_name) -> Result<Self, Self::Error> {
let json_value = serde_json::to_value(value)?;
Ok(match json_value {
serde_json::Value::Object(mut map) => {
map.remove("task_type");
serde_json::Value::Object(map)
}
_ => {
unreachable!("subscriber task must be an json object");
}
})
}
}
impl $task_enum_name {
pub fn task_type(&self) -> $type_enum_name {
match self {
@@ -121,6 +105,18 @@ macro_rules! register_subscriber_task_types {
}
}
fn set_subscriber_id(&mut self, subscriber_id: i32) {
match self {
$(Self::$task_variant(t) => t.set_subscriber_id(subscriber_id),)*
}
}
fn set_cron_id(&mut self, cron_id: Option<i32>) {
match self {
$(Self::$task_variant(t) => t.set_cron_id(cron_id),)*
}
}
fn from_input(input: Self::InputType, subscriber_id: i32) -> Self {
match input {
$(Self::InputType::$task_variant(t) =>
@@ -159,7 +155,7 @@ register_subscriber_task_types!(
}
},
task_enum: {
#[derive(Clone, Debug, PartialEq, Eq, FromJsonQueryResult)]
#[derive(Clone, Debug, PartialEq, FromJsonQueryResult)]
pub enum SubscriberTask {
SyncOneSubscriptionFeedsIncremental(SyncOneSubscriptionFeedsIncrementalTask),
SyncOneSubscriptionFeedsFull(SyncOneSubscriptionFeedsFullTask),

6
apps/recorder/src/task/registry/subscriber/subscription.rs
View File

@@ -39,7 +39,7 @@ macro_rules! register_subscription_task_type {
}
register_subscription_task_type! {
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq)]
pub struct SyncOneSubscriptionFeedsIncrementalTask {
} => async |subscription, ctx| -> RecorderResult<()> {
subscription.sync_feeds_incremental(ctx).await?;
@@ -48,7 +48,7 @@ register_subscription_task_type! {
}
register_subscription_task_type! {
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq)]
pub struct SyncOneSubscriptionFeedsFullTask {
} => async |subscription, ctx| -> RecorderResult<()> {
subscription.sync_feeds_full(ctx).await?;
@@ -57,7 +57,7 @@ register_subscription_task_type! {
}
register_subscription_task_type! {
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq)]
pub struct SyncOneSubscriptionSourcesTask {
} => async |subscription, ctx| -> RecorderResult<()> {
subscription.sync_sources(ctx).await?;

67
apps/recorder/src/task/registry/system/base.rs Normal file
View File

@@ -0,0 +1,67 @@
macro_rules! register_system_task_type {
(
$(#[$type_meta:meta])*
$task_vis:vis struct $task_name:ident {
$($(#[$field_meta:meta])* pub $field_name:ident: $field_type:ty),* $(,)?
}
) => {
$(#[$type_meta])*
#[derive(typed_builder::TypedBuilder, ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[ts(rename_all = "camelCase")]
$task_vis struct $task_name {
$($(#[$field_meta])* pub $field_name: $field_type,)*
#[serde(default, skip_serializing_if = "Option::is_none")]
#[builder(default = None)]
pub subscriber_id: Option<i32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[builder(default = None)]
pub cron_id: Option<i32>,
}
paste::paste! {
$(#[$type_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase")]
$task_vis struct [<$task_name Input>] {
$($(#[$field_meta])* pub $field_name: $field_type,)*
#[serde(default, skip_serializing_if = "Option::is_none")]
pub subscriber_id: Option<i32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cron_id: Option<i32>,
}
}
impl $crate::task::SystemTaskTrait for $task_name {
paste::paste! {
type InputType = [<$task_name Input>];
}
fn get_subscriber_id(&self) -> Option<i32> {
self.subscriber_id
}
fn get_cron_id(&self) -> Option<i32> {
self.cron_id
}
fn set_subscriber_id(&mut self, subscriber_id: Option<i32>) {
self.subscriber_id = subscriber_id;
}
fn set_cron_id(&mut self, cron_id: Option<i32>) {
self.cron_id = cron_id;
}
fn from_input(input: Self::InputType, subscriber_id: Option<i32>) -> Self {
Self {
$($field_name: input.$field_name,)*
subscriber_id: input.subscriber_id.or(subscriber_id),
cron_id: input.cron_id,
}
}
}
}
}
pub(crate) use register_system_task_type;

18
apps/recorder/src/task/registry/system/media.rs
View File

@@ -1,18 +1,22 @@
use std::sync::Arc;
use quirks_path::Path;
use serde::{Deserialize, Serialize};
use tracing::instrument;
use crate::{
app::AppContextTrait, errors::RecorderResult, media::EncodeImageOptions, task::AsyncTaskTrait,
app::AppContextTrait,
errors::RecorderResult,
media::EncodeImageOptions,
task::{AsyncTaskTrait, register_system_task_type},
};
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct OptimizeImageTask {
pub source_path: String,
pub target_path: String,
pub format_options: EncodeImageOptions,
register_system_task_type! {
#[derive(Clone, Debug, PartialEq)]
pub struct OptimizeImageTask {
pub source_path: String,
pub target_path: String,
pub format_options: EncodeImageOptions,
}
}
#[async_trait::async_trait]

128
apps/recorder/src/task/registry/system/mod.rs
View File

@@ -1,14 +1,15 @@
mod base;
mod media;
pub(crate) use base::register_system_task_type;
pub use media::OptimizeImageTask;
use sea_orm::{DeriveActiveEnum, DeriveDisplay, EnumIter, FromJsonQueryResult};
use serde::{Deserialize, Serialize};
macro_rules! register_system_task_types {
(
task_type_enum: {
$(#[$type_enum_meta:meta])*
pub enum $type_enum_name:ident {
$type_vis:vis enum $type_enum_name:ident {
$(
$(#[$variant_meta:meta])*
$variant:ident => $string_value:literal
@@ -17,16 +18,18 @@ macro_rules! register_system_task_types {
},
task_enum: {
$(#[$task_enum_meta:meta])*
pub enum $task_enum_name:ident {
$task_vis:vis enum $task_enum_name:ident {
$(
$(#[$task_variant_meta:meta])*
$task_variant:ident($task_type:ty)
),* $(,)?
}
}
) => {
$(#[$type_enum_meta])*
#[derive(serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[sea_orm(rs_type = "String", db_type = "Text")]
pub enum $type_enum_name {
$type_vis enum $type_enum_name {
$(
$(#[$variant_meta])*
#[serde(rename = $string_value)]
@@ -37,30 +40,17 @@ macro_rules! register_system_task_types {
$(#[$task_enum_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize, PartialEq)]
#[serde(tag = "task_type")]
pub enum $task_enum_name {
#[ts(export, rename = "SystemTaskType", rename_all = "camelCase", tag = "taskType")]
$task_vis enum $task_enum_name {
$(
$(#[$task_variant_meta])*
#[serde(rename = $string_value)]
$task_variant($task_type),
)*
}
impl TryFrom<$task_enum_name> for serde_json::Value {
type Error = $crate::errors::RecorderError;
fn try_from(value: $task_enum_name) -> Result<Self, Self::Error> {
let json_value = serde_json::to_value(value)?;
Ok(match json_value {
serde_json::Value::Object(mut map) => {
map.remove("task_type");
serde_json::Value::Object(map)
}
_ => {
unreachable!("subscriber task must be an json object");
}
})
}
}
impl $task_enum_name {
pub fn task_type(&self) -> $type_enum_name {
match self {
@@ -69,6 +59,21 @@ macro_rules! register_system_task_types {
}
}
paste::paste! {
$(#[$task_enum_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize, PartialEq)]
#[serde(tag = "taskType", rename_all = "camelCase")]
#[ts(export, rename_all = "camelCase", tag = "taskType")]
$task_vis enum [<$task_enum_name Input>] {
$(
$(#[$task_variant_meta])*
#[serde(rename = $string_value)]
$task_variant(<$task_type as $crate::task::SystemTaskTrait>::InputType),
)*
}
}
#[async_trait::async_trait]
impl $crate::task::AsyncTaskTrait for $task_enum_name {
async fn run_async(self, ctx: std::sync::Arc<dyn $crate::app::AppContextTrait>) -> $crate::errors::RecorderResult<()> {
@@ -78,18 +83,60 @@ macro_rules! register_system_task_types {
}
}
}
impl $crate::task::SystemTaskTrait for $task_enum_name {
paste::paste! {
type InputType = [<$task_enum_name Input>];
}
fn get_subscriber_id(&self) -> Option<i32> {
match self {
$(Self::$task_variant(t) => t.get_subscriber_id(),)*
}
}
fn get_cron_id(&self) -> Option<i32> {
match self {
$(Self::$task_variant(t) => t.get_cron_id(),)*
}
}
fn set_subscriber_id(&mut self, subscriber_id: Option<i32>) {
match self {
$(Self::$task_variant(t) => t.set_subscriber_id(subscriber_id),)*
}
}
fn set_cron_id(&mut self, cron_id: Option<i32>) {
match self {
$(Self::$task_variant(t) => t.set_cron_id(cron_id),)*
}
}
fn from_input(input: Self::InputType, subscriber_id: Option<i32>) -> Self {
match input {
$(Self::InputType::$task_variant(t) =>
Self::$task_variant(<$task_type as $crate::task::SystemTaskTrait>::from_input(t, subscriber_id)),)*
}
}
}
$(
impl From<$task_type> for $task_enum_name {
fn from(task: $task_type) -> Self {
Self::$task_variant(task)
}
}
)*
};
}
#[cfg(not(any(test, feature = "test-utils")))]
register_system_task_types! {
task_type_enum: {
#[derive(
Clone,
Debug,
Serialize,
Deserialize,
PartialEq,
Eq,
Copy,
DeriveActiveEnum,
DeriveDisplay,
@@ -100,9 +147,34 @@ register_system_task_types! {
}
},
task_enum: {
#[derive(Clone, Debug, Serialize, Deserialize, FromJsonQueryResult)]
#[derive(Clone, Debug, FromJsonQueryResult)]
pub enum SystemTask {
OptimizeImage(OptimizeImageTask),
OptimizeImage(OptimizeImageTask)
}
}
}
#[cfg(any(test, feature = "test-utils"))]
register_system_task_types! {
task_type_enum: {
#[derive(
Clone,
Debug,
Copy,
DeriveActiveEnum,
DeriveDisplay,
EnumIter
)]
pub enum SystemTaskType {
OptimizeImage => "optimize_image",
Test => "test",
}
},
task_enum: {
#[derive(Clone, Debug, FromJsonQueryResult)]
pub enum SystemTask {
OptimizeImage(OptimizeImageTask),
Test(crate::test_utils::task::TestSystemTask),
}
}
}

28
apps/recorder/src/task/service.rs
View File

@@ -294,3 +294,31 @@ impl TaskService {
}
}
}
#[cfg(test)]
#[allow(unused_variables)]
mod tests {
use rstest::{fixture, rstest};
use tracing::Level;
use super::*;
use crate::test_utils::{app::TestingPreset, tracing::try_init_testing_tracing};
#[fixture]
fn before_each() {
try_init_testing_tracing(Level::DEBUG);
}
#[rstest]
#[tokio::test]
async fn test_cron_due_listening(before_each: ()) -> RecorderResult<()> {
let mut preset = TestingPreset::default().await?;
let app_ctx = preset.app_ctx.clone();
let db = app_ctx.db();
todo!();
Ok(())
}
}

34
apps/recorder/src/test_utils/app.rs
View File

@@ -5,11 +5,12 @@ use typed_builder::TypedBuilder;
use crate::{
app::AppContextTrait,
errors::RecorderResult,
test_utils::{
crypto::build_testing_crypto_service,
database::{TestingDatabaseServiceConfig, build_testing_database_service},
media::build_testing_media_service,
mikan::build_testing_mikan_client,
mikan::{MikanMockServer, build_testing_mikan_client},
storage::build_testing_storage_service,
task::build_testing_task_service,
},
@@ -42,10 +43,8 @@ impl TestingAppContext {
self.task.get_or_init(|| task);
}
pub async fn from_preset(
preset: TestingAppContextPreset,
) -> crate::errors::RecorderResult<Arc<Self>> {
let mikan_client = build_testing_mikan_client(preset.mikan_base_url.clone()).await?;
pub async fn from_preset(preset: TestingAppContextPreset) -> RecorderResult<Arc<Self>> {
let mikan_client = build_testing_mikan_client(preset.mikan_base_url).await?;
let db_service =
build_testing_database_service(preset.database_config.unwrap_or_default()).await?;
let crypto_service = build_testing_crypto_service().await?;
@@ -137,3 +136,28 @@ pub struct TestingAppContextPreset {
pub mikan_base_url: String,
pub database_config: Option<TestingDatabaseServiceConfig>,
}
#[derive(TypedBuilder)]
pub struct TestingPreset {
pub mikan_server: MikanMockServer,
pub app_ctx: Arc<dyn AppContextTrait>,
}
impl TestingPreset {
pub async fn default() -> RecorderResult<Self> {
let mikan_server = MikanMockServer::new().await?;
let database_config = TestingDatabaseServiceConfig::default();
let app_ctx = TestingAppContext::from_preset(TestingAppContextPreset {
mikan_base_url: mikan_server.base_url().to_string(),
database_config: Some(database_config),
})
.await?;
let preset = Self::builder()
.mikan_server(mikan_server)
.app_ctx(app_ctx)
.build();
Ok(preset)
}
}

1
apps/recorder/src/test_utils/database.rs
View File

@@ -3,6 +3,7 @@ use crate::{
errors::RecorderResult,
};
#[derive(Clone, Debug)]
pub struct TestingDatabaseServiceConfig {
pub auto_migrate: bool,
}

21
apps/recorder/src/test_utils/mikan.rs
View File

@@ -1,5 +1,6 @@
use std::{
collections::HashMap,
fmt::Debug,
ops::{Deref, DerefMut},
path::{self, PathBuf},
};
@@ -148,13 +149,15 @@ impl AsRef<path::Path> for MikanDoppelPath {
}
}
#[cfg(any(test, debug_assertions, feature = "test-utils"))]
lazy_static! {
static ref TEST_RESOURCES_DIR: String =
if cfg!(any(test, debug_assertions, feature = "playground")) {
format!("{}/tests/resources", env!("CARGO_MANIFEST_DIR"))
} else {
"tests/resources".to_string()
};
format!("{}/tests/resources", env!("CARGO_MANIFEST_DIR"));
}
#[cfg(not(any(test, debug_assertions, feature = "test-utils")))]
lazy_static! {
static ref TEST_RESOURCES_DIR: String = "tests/resources".to_string();
}
impl From<Url> for MikanDoppelPath {
@@ -227,6 +230,14 @@ pub struct MikanMockServer {
base_url: Url,
}
impl Debug for MikanMockServer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("MikanMockServer")
.field("base_url", &self.base_url)
.finish()
}
}
impl MikanMockServer {
pub async fn new_with_port(port: u16) -> RecorderResult<Self> {
let server = mockito::Server::new_with_opts_async(mockito::ServerOpts {

30
apps/recorder/src/test_utils/task.rs
View File

@@ -1,15 +1,43 @@
use std::sync::Arc;
use chrono::Utc;
use crate::{
app::AppContextTrait,
errors::RecorderResult,
task::{TaskConfig, TaskService},
task::{AsyncTaskTrait, TaskConfig, TaskService, register_system_task_type},
};
register_system_task_type! {
#[derive(Debug, Clone, PartialEq)]
pub struct TestSystemTask {
pub task_id: String,
}
}
#[async_trait::async_trait]
impl AsyncTaskTrait for TestSystemTask {
async fn run_async(self, ctx: Arc<dyn AppContextTrait>) -> RecorderResult<()> {
let storage = ctx.storage();
storage
.write(
storage.build_test_path(self.task_id),
serde_json::json!({ "exec_time": Utc::now().timestamp_millis() })
.to_string()
.into(),
)
.await?;
Ok(())
}
}
pub async fn build_testing_task_service(
ctx: Arc<dyn AppContextTrait>,
) -> RecorderResult<TaskService> {
let config = TaskConfig::default();
let task_service = TaskService::from_config_and_ctx(config, ctx).await?;
Ok(task_service)
}