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

refactor: refactor graphql

Browse Source
This commit is contained in:
master
2025-07-02 01:25:44 +08:00
parent d06acde882
commit 5b001f9584
34 changed files with 376 additions and 623 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
apps/recorder/Cargo.toml
View File

@@ -125,6 +125,7 @@ seaography = { version = "1.1", features = [
"with-bigdecimal",
"with-postgres-array",
"with-json-as-scalar",
"with-custom-as-json",
] }
tower = { version = "0.5.2", features = ["util"] }
tower-http = { version = "0.6", features = [
@@ -167,8 +168,7 @@ quick-xml = { version = "0.37.5", features = [
croner = "2.2.0"
ts-rs = "11.0.1"
secrecy = { version = "0.10.3", features = ["serde"] }
schemars = "1.0.3"
jsonschema = "0.30.0"
paste = "1.0.15"
[dev-dependencies]
inquire = { workspace = true }

6
apps/recorder/bindings/SubscriberTaskInput.ts Normal file
View File

@@ -0,0 +1,6 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { SyncOneSubscriptionFeedsFullTaskInput } from "./SyncOneSubscriptionFeedsFullTaskInput";
import type { SyncOneSubscriptionFeedsIncrementalTaskInput } from "./SyncOneSubscriptionFeedsIncrementalTaskInput";
import type { SyncOneSubscriptionSourcesTaskInput } from "./SyncOneSubscriptionSourcesTaskInput";
export type SubscriberTaskInput = { "taskType": "sync_one_subscription_feeds_incremental" } & SyncOneSubscriptionFeedsIncrementalTaskInput | { "taskType": "sync_one_subscription_feeds_full" } & SyncOneSubscriptionFeedsFullTaskInput | { "taskType": "sync_one_subscription_sources" } & SyncOneSubscriptionSourcesTaskInput;

2
apps/recorder/bindings/SubscriberTask.ts → apps/recorder/bindings/SubscriberTaskType.ts
View File

@@ -3,4 +3,4 @@ import type { SyncOneSubscriptionFeedsFullTask } from "./SyncOneSubscriptionFeed
import type { SyncOneSubscriptionFeedsIncrementalTask } from "./SyncOneSubscriptionFeedsIncrementalTask";
import type { SyncOneSubscriptionSourcesTask } from "./SyncOneSubscriptionSourcesTask";
export type SubscriberTask = { "taskType": "sync_one_subscription_feeds_incremental" } & SyncOneSubscriptionFeedsIncrementalTask | { "taskType": "sync_one_subscription_feeds_full" } & SyncOneSubscriptionFeedsFullTask | { "taskType": "sync_one_subscription_sources" } & SyncOneSubscriptionSourcesTask;
export type SubscriberTaskType = { "taskType": "sync_one_subscription_feeds_incremental" } & SyncOneSubscriptionFeedsIncrementalTask | { "taskType": "sync_one_subscription_feeds_full" } & SyncOneSubscriptionFeedsFullTask | { "taskType": "sync_one_subscription_sources" } & SyncOneSubscriptionSourcesTask;

2
apps/recorder/bindings/SyncOneSubscriptionFeedsFullTask.ts
View File

@@ -1,3 +1,3 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type SyncOneSubscriptionFeedsFullTask = { subscriptionId: number, subscriberId: number, cronId: number | null, };
export type SyncOneSubscriptionFeedsFullTask = { subscriptionId: number, subscriberId: number, cronId?: number | null, };

3
apps/recorder/bindings/SyncOneSubscriptionFeedsFullTaskInput.ts Normal file
View File

@@ -0,0 +1,3 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type SyncOneSubscriptionFeedsFullTaskInput = { subscriptionId: number, subscriberId?: number | null, cronId?: number | null, };

2
apps/recorder/bindings/SyncOneSubscriptionFeedsIncrementalTask.ts
View File

@@ -1,3 +1,3 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type SyncOneSubscriptionFeedsIncrementalTask = { subscriptionId: number, subscriberId: number, cronId: number | null, };
export type SyncOneSubscriptionFeedsIncrementalTask = { subscriptionId: number, subscriberId: number, cronId?: number | null, };

3
apps/recorder/bindings/SyncOneSubscriptionFeedsIncrementalTaskInput.ts Normal file
View File

@@ -0,0 +1,3 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type SyncOneSubscriptionFeedsIncrementalTaskInput = { subscriptionId: number, subscriberId?: number | null, cronId?: number | null, };

2
apps/recorder/bindings/SyncOneSubscriptionSourcesTask.ts
View File

@@ -1,3 +1,3 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type SyncOneSubscriptionSourcesTask = { subscriptionId: number, subscriberId: number, cronId: number | null, };
export type SyncOneSubscriptionSourcesTask = { subscriptionId: number, subscriberId: number, cronId?: number | null, };

3
apps/recorder/bindings/SyncOneSubscriptionSourcesTaskInput.ts Normal file
View File

@@ -0,0 +1,3 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type SyncOneSubscriptionSourcesTaskInput = { subscriptionId: number, subscriberId?: number | null, cronId?: number | null, };

4
apps/recorder/src/graphql/domains/credential_3rd.rs
View File

@@ -105,10 +105,10 @@ pub fn register_credential3rd_to_schema_builder(
{
let check_available_mutation_name = get_entity_custom_mutation_field_name::<
credential_3rd::Entity,
>(&builder_context, "CheckAvailable");
>(builder_context, "CheckAvailable");
let check_available_mutation =
generate_entity_filtered_mutation_field::<credential_3rd::Entity, _, _>(
builder_context.clone(),
builder_context,
check_available_mutation_name,
TypeRef::named_nn(Credential3rdCheckAvailableInfo::object_type_name()),
Arc::new(|_resolver_ctx, app_ctx, filters| {

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

@@ -1,4 +1,3 @@
use convert_case::Case;
use sea_orm::Iterable;
use seaography::{Builder as SeaographyBuilder, BuilderContext};
@@ -44,11 +43,7 @@ fn skip_columns_for_entity_input(context: &mut BuilderContext) {
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,
Some(Case::Snake),
);
restrict_subscriber_tasks_for_entity::<cron::Entity>(context, &cron::Column::SubscriberTask);
skip_columns_for_entity_input(context);
}

2
apps/recorder/src/graphql/domains/feeds.rs
View File

@@ -29,7 +29,7 @@ pub fn register_feeds_to_schema_context(context: &mut BuilderContext) {
context.types.input_none_conversions.insert(
get_entity_and_column_name::<feeds::Entity>(context, &feeds::Column::Token),
Arc::new(
Box::new(
move |context: &ResolverContext| -> SeaResult<Option<SeaValue>> {
let field_name = context.field().name();
if field_name == entity_create_one_mutation_field_name.as_str()

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

@@ -1,6 +1,6 @@
use std::{ops::Deref, sync::Arc};
use async_graphql::dynamic::{FieldValue, TypeRef};
use async_graphql::dynamic::{FieldValue, Scalar, TypeRef};
use convert_case::Case;
use sea_orm::{
ActiveModelBehavior, ColumnTrait, ConnectionTrait, EntityTrait, Iterable, QueryFilter,
@@ -9,9 +9,9 @@ use sea_orm::{
use seaography::{
Builder as SeaographyBuilder, BuilderContext, SeaographyError, prepare_active_model,
};
use ts_rs::TS;
use crate::{
app::AppContextTrait,
auth::AuthUserInfo,
errors::RecorderError,
graphql::{
@@ -32,7 +32,6 @@ use crate::{
},
models::subscriber_tasks,
task::{ApalisJobs, ApalisSchema, SubscriberTaskTrait},
utils::json::convert_json_keys,
};
fn skip_columns_for_entity_input(context: &mut BuilderContext) {
@@ -49,11 +48,8 @@ fn skip_columns_for_entity_input(context: &mut BuilderContext) {
}
}
pub fn restrict_subscriber_tasks_for_entity<T>(
context: &mut BuilderContext,
column: &T::Column,
case: Option<Case<'static>>,
) where
pub fn restrict_subscriber_tasks_for_entity<T>(context: &mut BuilderContext, column: &T::Column)
where
T: EntityTrait,
<T as EntityTrait>::Model: Sync,
{
@@ -62,42 +58,33 @@ pub fn restrict_subscriber_tasks_for_entity<T>(
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(subscriber_tasks::SubscriberTask::ident().into()),
);
context.types.output_type_overwrites.insert(
entity_column_name.clone(),
TypeRef::Named(subscriber_tasks::SubscriberTask::ident().into()),
);
context.types.input_conversions.insert(
entity_column_name.clone(),
Arc::new(move |resolve_context, value_accessor| {
let mut json_value = value_accessor
.as_value()
.clone()
.into_json()
.map_err(|err| {
SeaographyError::TypeConversionError(
err.to_string(),
format!("Json - {entity_column_name}"),
)
})?;
if let Some(case) = case {
json_value = convert_json_keys(json_value, case);
}
Box::new(move |resolve_context, value_accessor| {
let task: subscriber_tasks::SubscriberTaskInput = value_accessor.deserialize()?;
let subscriber_id = resolve_context
.data::<AuthUserInfo>()?
.subscriber_auth
.subscriber_id;
if let Some(obj) = json_value.as_object_mut() {
obj.entry("subscriber_id")
.or_insert_with(|| serde_json::Value::from(subscriber_id));
}
let task = subscriber_tasks::SubscriberTask::from_input(task, subscriber_id);
subscriber_tasks::subscriber_task_schema()
.validate(&json_value)
.map_err(|err| {
SeaographyError::TypeConversionError(
err.to_string(),
format!("Json - {entity_column_name}"),
)
})?;
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))))
}),
@@ -114,7 +101,6 @@ pub fn register_subscriber_tasks_to_schema_context(context: &mut BuilderContext)
restrict_subscriber_tasks_for_entity::<subscriber_tasks::Entity>(
context,
&subscriber_tasks::Column::Job,
Some(Case::Snake),
);
skip_columns_for_entity_input(context);
@@ -123,22 +109,26 @@ pub fn register_subscriber_tasks_to_schema_context(context: &mut BuilderContext)
pub fn register_subscriber_tasks_to_schema_builder(
mut builder: SeaographyBuilder,
) -> SeaographyBuilder {
builder.schema = builder.schema.register(
Scalar::new(subscriber_tasks::SubscriberTask::ident())
.description(subscriber_tasks::SubscriberTask::decl()),
);
builder.register_enumeration::<subscriber_tasks::SubscriberTaskType>();
builder.register_enumeration::<subscriber_tasks::SubscriberTaskStatus>();
builder = register_entity_default_readonly!(builder, subscriber_tasks);
let builder_context = builder.context.clone();
let builder_context = builder.context;
{
builder
.outputs
.push(generate_entity_default_basic_entity_object::<
subscriber_tasks::Entity,
>(builder_context.clone()));
>(builder_context));
}
{
let delete_mutation = generate_entity_delete_mutation_field::<subscriber_tasks::Entity>(
builder_context.clone(),
builder_context,
Arc::new(|_resolver_ctx, app_ctx, filters| {
Box::pin(async move {
let db = app_ctx.db();
@@ -169,13 +159,13 @@ pub fn register_subscriber_tasks_to_schema_builder(
{
let entity_retry_one_mutation_name = get_entity_custom_mutation_field_name::<
subscriber_tasks::Entity,
>(&builder_context, "RetryOne");
>(builder_context, "RetryOne");
let retry_one_mutation =
generate_entity_filtered_mutation_field::<subscriber_tasks::Entity, _, _>(
builder_context.clone(),
builder_context,
entity_retry_one_mutation_name,
TypeRef::named_nn(get_entity_basic_type_name::<subscriber_tasks::Entity>(
&builder_context,
builder_context,
)),
Arc::new(|_resolver_ctx, app_ctx, filters| {
Box::pin(async move {
@@ -214,15 +204,15 @@ pub fn register_subscriber_tasks_to_schema_builder(
.inputs
.push(generate_entity_default_insert_input_object::<
subscriber_tasks::Entity,
>(&builder.context));
>(builder_context));
let create_one_mutation =
generate_entity_create_one_mutation_field::<subscriber_tasks::Entity>(
builder.context.clone(),
builder_context,
Arc::new(move |resolver_ctx, app_ctx, input_object| {
let active_model: Result<subscriber_tasks::ActiveModel, _> =
prepare_active_model(&builder_context.clone(), &input_object, resolver_ctx);
Box::pin(async move {
let active_model: Result<subscriber_tasks::ActiveModel, _> =
prepare_active_model(builder_context, &input_object, resolver_ctx);
let task_service = app_ctx.task();
let active_model = active_model?;

6
apps/recorder/src/graphql/domains/subscribers.rs
View File

@@ -79,7 +79,7 @@ where
T: EntityTrait,
<T as EntityTrait>::Model: Sync,
{
Arc::new(move |context: &ResolverContext| -> GuardAction {
Box::new(move |context: &ResolverContext| -> GuardAction {
match context.ctx.data::<AuthUserInfo>() {
Ok(_) => GuardAction::Allow,
Err(err) => GuardAction::Block(Some(err.message)),
@@ -106,7 +106,7 @@ where
let entity_update_mutation_data_field_name =
Arc::new(get_entity_update_mutation_data_field_name(context).to_string());
Arc::new(move |context: &ResolverContext| -> GuardAction {
Box::new(move |context: &ResolverContext| -> GuardAction {
match context.ctx.data::<AuthUserInfo>() {
Ok(user_info) => {
let subscriber_id = user_info.subscriber_auth.subscriber_id;
@@ -253,7 +253,7 @@ where
Arc::new(get_entity_create_one_mutation_field_name::<T>(context));
let entity_create_batch_mutation_field_name =
Arc::new(get_entity_create_batch_mutation_field_name::<T>(context));
Arc::new(
Box::new(
move |context: &ResolverContext| -> SeaResult<Option<SeaValue>> {
let field_name = context.field().name();
if field_name == entity_create_one_mutation_field_name.as_str()

2
apps/recorder/src/graphql/domains/subscriptions.rs
View File

@@ -2,7 +2,7 @@ use seaography::{Builder as SeaographyBuilder, BuilderContext};
use crate::{
graphql::{
domains::subscribers::restrict_subscriber_for_entity, infra,
domains::subscribers::restrict_subscriber_for_entity,
infra::custom::register_entity_default_writable,
},
models::subscriptions,

4
apps/recorder/src/graphql/infra/crypto.rs
View File

@@ -16,7 +16,7 @@ pub fn register_crypto_column_input_conversion_to_schema_context<T>(
{
context.types.input_conversions.insert(
get_entity_and_column_name::<T>(context, column),
Arc::new(
Box::new(
move |_resolve_context: &ResolverContext<'_>,
value: &ValueAccessor|
-> SeaResult<sea_orm::Value> {
@@ -38,7 +38,7 @@ pub fn register_crypto_column_output_conversion_to_schema_context<T>(
{
context.types.output_conversions.insert(
get_entity_and_column_name::<T>(context, column),
Arc::new(
Box::new(
move |value: &sea_orm::Value| -> SeaResult<async_graphql::Value> {
if let SeaValue::String(s) = value {
if let Some(s) = s {

267
apps/recorder/src/graphql/infra/custom.rs
View File

@@ -34,9 +34,9 @@ pub type FilterMutationFn = Arc<
pub type CreateOneMutationFn<M> = Arc<
dyn for<'a> Fn(
&ResolverContext<'a>,
&'a ResolverContext<'a>,
Arc<dyn AppContextTrait>,
ObjectAccessor<'_>,
ObjectAccessor<'a>,
) -> Pin<Box<dyn Future<Output = RecorderResult<M>> + Send + 'a>>
+ Send
+ Sync,
@@ -44,9 +44,9 @@ pub type CreateOneMutationFn<M> = Arc<
pub type CreateBatchMutationFn<M> = Arc<
dyn for<'a> Fn(
&ResolverContext<'a>,
&'a ResolverContext<'a>,
Arc<dyn AppContextTrait>,
Vec<ObjectAccessor<'_>>,
Vec<ObjectAccessor<'a>>,
) -> Pin<Box<dyn Future<Output = RecorderResult<Vec<M>>> + Send + 'a>>
+ Send
+ Sync,
@@ -54,10 +54,10 @@ pub type CreateBatchMutationFn<M> = Arc<
pub type UpdateMutationFn<M> = Arc<
dyn for<'a> Fn(
&ResolverContext<'a>,
&'a ResolverContext<'a>,
Arc<dyn AppContextTrait>,
Condition,
ObjectAccessor<'_>,
ObjectAccessor<'a>,
) -> Pin<Box<dyn Future<Output = RecorderResult<Vec<M>>> + Send + 'a>>
+ Send
+ Sync,
@@ -89,12 +89,13 @@ where
EntityInputBuilder::update_input_object::<T>(context)
}
pub fn generate_entity_default_basic_entity_object<T>(context: Arc<BuilderContext>) -> Object
pub fn generate_entity_default_basic_entity_object<T>(context: &'static BuilderContext) -> Object
where
T: EntityTrait,
<T as EntityTrait>::Model: Sync,
{
EntityObjectBuilder::basic_to_object::<T>(context)
let entity_object_builder = EntityObjectBuilder { context };
entity_object_builder.basic_to_object::<T>()
}
pub fn generate_entity_input_object<T>(
@@ -113,7 +114,7 @@ where
}
pub fn generate_entity_filtered_mutation_field<E, N, R>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
field_name: N,
type_ref: R,
mutation_fn: FilterMutationFn,
@@ -124,189 +125,221 @@ where
N: Into<String>,
R: Into<TypeRef>,
{
let object_name: String = get_entity_name::<E>(&builder_context);
let object_name: String = get_entity_name::<E>(builder_context);
let guard = builder_context
.guards
.entity_guards
.get(&object_name)
.cloned();
let guard = builder_context.guards.entity_guards.get(&object_name);
let filter_input_value = InputValue::new(
get_entity_renormalized_filter_field_name(),
TypeRef::named(get_entity_filter_input_type_name::<E>(&builder_context)),
);
Field::new(field_name, type_ref, move |ctx| {
Field::new(field_name, type_ref, move |resolve_context| {
let mutation_fn = mutation_fn.clone();
let builder_context = builder_context.clone();
let guard_flag = if let Some(guard) = guard.as_ref() {
(*guard)(&ctx)
} else {
GuardAction::Allow
};
FieldFuture::new(async move {
let guard_flag = if let Some(guard) = guard {
(*guard)(&resolve_context)
} else {
GuardAction::Allow
};
if let GuardAction::Block(reason) = guard_flag {
return Err::<Option<_>, async_graphql::Error>(async_graphql::Error::new(
reason.unwrap_or("Entity guard triggered.".into()),
));
}
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let filters = resolve_context
.args
.get(get_entity_renormalized_filter_field_name());
let filters = ctx.args.get(get_entity_renormalized_filter_field_name());
let filters = get_filter_conditions::<E>(&resolve_context, builder_context, filters);
let filters = get_filter_conditions::<E>(&ctx, &builder_context, filters);
let app_ctx = resolve_context.data::<Arc<dyn AppContextTrait>>()?;
let result = mutation_fn(&ctx, app_ctx.clone(), filters)
.await
.map_err(async_graphql::Error::new_with_source)?;
let result = mutation_fn(&resolve_context, app_ctx.clone(), filters).await?;
Ok(result)
})
})
.argument(filter_input_value)
.argument(InputValue::new(
get_entity_renormalized_filter_field_name(),
TypeRef::named(get_entity_filter_input_type_name::<E>(builder_context)),
))
}
pub fn generate_entity_create_one_mutation_field<E>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
mutation_fn: CreateOneMutationFn<E::Model>,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
{
EntityCreateOneMutationBuilder::to_field_with_mutation_fn::<E>(
builder_context.clone(),
Arc::new(move |resolver_ctx, input_object| {
let result = resolver_ctx
.data::<Arc<dyn AppContextTrait>>()
.map(|app_ctx| mutation_fn(&resolver_ctx, app_ctx.clone(), input_object));
Box::pin(async move { result?.await.map_err(async_graphql::Error::new_with_source) })
}),
)
let entity_create_one_mutation_builder = EntityCreateOneMutationBuilder {
context: builder_context,
};
entity_create_one_mutation_builder.to_field_with_mutation_fn::<E>(Arc::new(
move |resolver_ctx, input_object| {
let mutation_fn = mutation_fn.clone();
Box::pin(async move {
let app_ctx = resolver_ctx.data::<Arc<dyn AppContextTrait>>()?;
let result = mutation_fn(resolver_ctx, app_ctx.clone(), input_object).await?;
Ok(result)
})
},
))
}
pub fn generate_entity_default_create_one_mutation_field<E, A>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
active_model_hooks: bool,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
<E as EntityTrait>::Model: IntoActiveModel<A>,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send + 'static,
<E as EntityTrait>::Model: Sync + IntoActiveModel<A>,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send,
{
EntityCreateOneMutationBuilder::to_field::<E, A>(builder_context, active_model_hooks)
let entity_create_one_mutation_builder = EntityCreateOneMutationBuilder {
context: builder_context,
};
entity_create_one_mutation_builder.to_field::<E, A>(active_model_hooks)
}
pub fn generate_entity_create_batch_mutation_field<E, ID>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
mutation_fn: CreateBatchMutationFn<E::Model>,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
{
EntityCreateBatchMutationBuilder::to_field_with_mutation_fn::<E>(
builder_context,
Arc::new(move |resolver_ctx, input_objects| {
let result = resolver_ctx
.data::<Arc<dyn AppContextTrait>>()
.map(|app_ctx| mutation_fn(&resolver_ctx, app_ctx.clone(), input_objects));
Box::pin(async move { result?.await.map_err(async_graphql::Error::new_with_source) })
}),
)
let entity_create_batch_mutation_builder = EntityCreateBatchMutationBuilder {
context: builder_context,
};
entity_create_batch_mutation_builder.to_field_with_mutation_fn::<E>(Arc::new(
move |resolver_ctx, input_objects| {
let mutation_fn = mutation_fn.clone();
Box::pin(async move {
let app_ctx = resolver_ctx.data::<Arc<dyn AppContextTrait>>()?;
let result = mutation_fn(resolver_ctx, app_ctx.clone(), input_objects).await?;
Ok(result)
})
},
))
}
pub fn generate_entity_default_create_batch_mutation_field<E, A>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
active_model_hooks: bool,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
<E as EntityTrait>::Model: IntoActiveModel<A>,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send + 'static,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send,
{
EntityCreateBatchMutationBuilder::to_field::<E, A>(builder_context, active_model_hooks)
let entity_create_batch_mutation_builder = EntityCreateBatchMutationBuilder {
context: builder_context,
};
entity_create_batch_mutation_builder.to_field::<E, A>(active_model_hooks)
}
pub fn generate_entity_update_mutation_field<E>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
mutation_fn: UpdateMutationFn<E::Model>,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
{
EntityUpdateMutationBuilder::to_field_with_mutation_fn::<E>(
builder_context.clone(),
Arc::new(move |resolver_ctx, filters, input_object| {
let result = resolver_ctx
.data::<Arc<dyn AppContextTrait>>()
.map(|app_ctx| {
mutation_fn(
&resolver_ctx,
app_ctx.clone(),
get_filter_conditions::<E>(&resolver_ctx, &builder_context, filters),
input_object,
)
});
Box::pin(async move { result?.await.map_err(async_graphql::Error::new_with_source) })
}),
)
let entity_update_mutation_builder = EntityUpdateMutationBuilder {
context: builder_context,
};
entity_update_mutation_builder.to_field_with_mutation_fn::<E>(Arc::new(
move |resolver_ctx, filters, input_object| {
let mutation_fn = mutation_fn.clone();
Box::pin(async move {
let app_ctx = resolver_ctx.data::<Arc<dyn AppContextTrait>>()?;
let result = mutation_fn(
resolver_ctx,
app_ctx.clone(),
get_filter_conditions::<E>(resolver_ctx, builder_context, filters),
input_object,
)
.await
.map_err(async_graphql::Error::new_with_source)?;
Ok(result)
})
},
))
}
pub fn generate_entity_default_update_mutation_field<E, A>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
active_model_hooks: bool,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
<E as EntityTrait>::Model: IntoActiveModel<A>,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send + 'static,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send,
{
EntityUpdateMutationBuilder::to_field::<E, A>(builder_context, active_model_hooks)
let entity_update_mutation_builder = EntityUpdateMutationBuilder {
context: builder_context,
};
entity_update_mutation_builder.to_field::<E, A>(active_model_hooks)
}
pub fn generate_entity_delete_mutation_field<E>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
mutation_fn: DeleteMutationFn,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
{
EntityDeleteMutationBuilder::to_field_with_mutation_fn::<E>(
builder_context.clone(),
Arc::new(move |resolver_ctx, filters| {
let result = resolver_ctx
.data::<Arc<dyn AppContextTrait>>()
.map(|app_ctx| {
mutation_fn(
&resolver_ctx,
app_ctx.clone(),
get_filter_conditions::<E>(&resolver_ctx, &builder_context, filters),
)
});
Box::pin(async move { result?.await.map_err(async_graphql::Error::new_with_source) })
}),
)
let entity_delete_mutation_builder = EntityDeleteMutationBuilder {
context: builder_context,
};
entity_delete_mutation_builder.to_field_with_mutation_fn::<E>(Arc::new(
move |resolver_ctx, filters| {
let mutation_fn = mutation_fn.clone();
Box::pin(async move {
let app_ctx = resolver_ctx.data::<Arc<dyn AppContextTrait>>()?;
let result = mutation_fn(
resolver_ctx,
app_ctx.clone(),
get_filter_conditions::<E>(resolver_ctx, builder_context, filters),
)
.await
.map_err(async_graphql::Error::new_with_source)?;
Ok(result)
})
},
))
}
pub fn generate_entity_default_delete_mutation_field<E, A>(
builder_context: Arc<BuilderContext>,
builder_context: &'static BuilderContext,
active_model_hooks: bool,
) -> Field
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync,
<E as EntityTrait>::Model: IntoActiveModel<A>,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send + 'static,
<E as EntityTrait>::Model: Sync + IntoActiveModel<A>,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send,
{
EntityDeleteMutationBuilder::to_field::<E, A>(builder_context, active_model_hooks)
let entity_delete_mutation_builder = EntityDeleteMutationBuilder {
context: builder_context,
};
entity_delete_mutation_builder.to_field::<E, A>(active_model_hooks)
}
pub fn register_entity_default_mutations<E, A>(
@@ -316,37 +349,31 @@ pub fn register_entity_default_mutations<E, A>(
where
E: EntityTrait,
<E as EntityTrait>::Model: Sync + IntoActiveModel<A>,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send + 'static,
A: ActiveModelTrait<Entity = E> + sea_orm::ActiveModelBehavior + std::marker::Send,
{
let builder_context = &builder.context;
let builder_context = builder.context;
builder
.outputs
.push(generate_entity_default_basic_entity_object::<E>(
builder_context.clone(),
builder_context,
));
builder.inputs.extend([
generate_entity_default_insert_input_object::<E>(&builder.context),
generate_entity_default_update_input_object::<E>(&builder.context),
generate_entity_default_insert_input_object::<E>(builder_context),
generate_entity_default_update_input_object::<E>(builder_context),
]);
builder.mutations.extend([
generate_entity_default_create_one_mutation_field::<E, A>(
builder_context.clone(),
builder_context,
active_model_hooks,
),
generate_entity_default_create_batch_mutation_field::<E, A>(
builder_context.clone(),
active_model_hooks,
),
generate_entity_default_update_mutation_field::<E, A>(
builder_context.clone(),
active_model_hooks,
),
generate_entity_default_delete_mutation_field::<E, A>(
builder_context.clone(),
builder_context,
active_model_hooks,
),
generate_entity_default_update_mutation_field::<E, A>(builder_context, active_model_hooks),
generate_entity_default_delete_mutation_field::<E, A>(builder_context, active_model_hooks),
]);
builder
@@ -364,7 +391,7 @@ where
{
builder.register_entity::<T>(
<RE as sea_orm::Iterable>::iter()
.map(|rel| RelationBuilder::get_relation(&rel, builder.context.clone()))
.map(|rel| RelationBuilder::get_relation(&rel, builder.context))
.collect(),
);
builder = builder.register_entity_dataloader_one_to_one(entity, tokio::spawn);
@@ -380,7 +407,7 @@ pub(crate) fn register_entity_default_writable_impl<T, RE, A, I>(
where
T: EntityTrait,
<T as EntityTrait>::Model: Sync + IntoActiveModel<A>,
A: ActiveModelTrait<Entity = T> + sea_orm::ActiveModelBehavior + std::marker::Send + 'static,
A: ActiveModelTrait<Entity = T> + sea_orm::ActiveModelBehavior + std::marker::Send,
RE: sea_orm::Iterable<Iterator = I> + RelationBuilder,
I: Iterator<Item = RE> + Clone + DoubleEndedIterator + ExactSizeIterator + FusedIterator,
{

30
apps/recorder/src/graphql/infra/json.rs
View File

@@ -1,5 +1,3 @@
use std::sync::Arc;
use async_graphql::{
Error as GraphqlError,
dynamic::{ResolverContext, Scalar, SchemaError},
@@ -7,7 +5,6 @@ use async_graphql::{
};
use convert_case::Case;
use itertools::Itertools;
use jsonschema::Validator;
use rust_decimal::{Decimal, prelude::FromPrimitive};
use sea_orm::{
Condition, EntityTrait,
@@ -946,16 +943,20 @@ where
T: EntityTrait,
<T as EntityTrait>::Model: Sync,
{
let entity_column_name = get_entity_and_column_name::<T>(context, column);
context.filter_types.overwrites.insert(
get_entity_and_column_name::<T>(context, column),
Some(FilterType::Custom(JSONB_FILTER_NAME.to_string())),
);
context.filter_types.condition_functions.insert(
entity_column_name.clone(),
generate_jsonb_filter_condition_function::<T>(context, column),
);
}
pub fn try_convert_jsonb_input_for_entity<T, S>(
context: &mut BuilderContext,
column: &T::Column,
validator: &'static Validator,
case: Option<Case<'static>>,
) where
T: EntityTrait,
@@ -965,19 +966,14 @@ pub fn try_convert_jsonb_input_for_entity<T, S>(
let entity_column_name = get_entity_and_column_name::<T>(context, column);
context.types.input_conversions.insert(
entity_column_name.clone(),
Arc::new(move |_resolve_context, accessor| {
let mut json_value = accessor.as_value().clone().into_json().map_err(|err| {
SeaographyError::TypeConversionError(
err.to_string(),
format!("Json - {entity_column_name}"),
)
})?;
Box::new(move |_resolve_context, accessor| {
let mut json_value: serde_json::Value = accessor.deserialize()?;
if let Some(case) = case {
json_value = convert_json_keys(json_value, case);
}
validator.validate(&json_value).map_err(|err| {
serde_json::from_value::<S>(json_value.clone()).map_err(|err| {
SeaographyError::TypeConversionError(
err.to_string(),
format!("Json - {entity_column_name}"),
@@ -997,10 +993,10 @@ pub fn convert_jsonb_output_for_entity<T>(
T: EntityTrait,
<T as EntityTrait>::Model: Sync,
{
let entity_column_key = get_entity_and_column_name::<T>(context, column);
let entity_column_name = get_entity_and_column_name::<T>(context, column);
context.types.output_conversions.insert(
entity_column_key.clone(),
Arc::new(move |value| {
entity_column_name.clone(),
Box::new(move |value| {
if let sea_orm::Value::Json(Some(json)) = value {
let mut json_value = json.as_ref().clone();
if let Some(case) = case {
@@ -1009,14 +1005,14 @@ pub fn convert_jsonb_output_for_entity<T>(
let result = async_graphql::Value::from_json(json_value).map_err(|err| {
SeaographyError::TypeConversionError(
err.to_string(),
format!("Json - {entity_column_key}"),
format!("Json - {entity_column_name}"),
)
})?;
Ok(result)
} else {
Err(SeaographyError::TypeConversionError(
"value should be json".to_string(),
format!("Json - {entity_column_key}"),
format!("Json - {entity_column_name}"),
))
}
}),

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

@@ -59,24 +59,27 @@ pub fn build_schema(
) -> Result<Schema, SchemaError> {
let database = app_ctx.db().as_ref().clone();
let context = Arc::new({
let context = CONTEXT.get_or_init(|| {
let mut context = BuilderContext::default();
// basic
renormalize_filter_field_names_to_schema_context(&mut context);
renormalize_data_field_names_to_schema_context(&mut context);
// domains
register_feeds_to_schema_context(&mut context);
register_subscribers_to_schema_context(&mut context);
register_subscriptions_to_schema_context(&mut context);
register_subscriber_tasks_to_schema_context(&mut context);
register_credential3rd_to_schema_context(&mut context, app_ctx.clone());
register_downloaders_to_schema_context(&mut context);
register_downloads_to_schema_context(&mut context);
register_episodes_to_schema_context(&mut context);
register_subscription_bangumi_to_schema_context(&mut context);
register_subscription_episode_to_schema_context(&mut context);
register_bangumi_to_schema_context(&mut context);
register_cron_to_schema_context(&mut context);
{
// domains
register_feeds_to_schema_context(&mut context);
register_subscribers_to_schema_context(&mut context);
register_subscriptions_to_schema_context(&mut context);
register_subscriber_tasks_to_schema_context(&mut context);
register_credential3rd_to_schema_context(&mut context, app_ctx.clone());
register_downloaders_to_schema_context(&mut context);
register_downloads_to_schema_context(&mut context);
register_episodes_to_schema_context(&mut context);
register_subscription_bangumi_to_schema_context(&mut context);
register_subscription_episode_to_schema_context(&mut context);
register_bangumi_to_schema_context(&mut context);
register_cron_to_schema_context(&mut context);
}
context
});

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

@@ -3,8 +3,8 @@ use sea_orm::{ActiveValue, entity::prelude::*};
use crate::task::SubscriberTaskTrait;
pub use crate::task::{
SubscriberTask, SubscriberTaskType, SubscriberTaskTypeEnum, SubscriberTaskTypeVariant,
SubscriberTaskTypeVariantIter, subscriber_task_schema,
SubscriberTask, SubscriberTaskInput, SubscriberTaskType, SubscriberTaskTypeEnum,
SubscriberTaskTypeVariant, SubscriberTaskTypeVariantIter,
};
#[derive(Clone, Debug, PartialEq, Eq, DeriveActiveEnum, EnumIter, DeriveDisplay)]

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

@@ -42,9 +42,13 @@ where
}
pub trait SubscriberTaskTrait: AsyncTaskTrait {
type InputType: Serialize + DeserializeOwned + Sized + Send;
fn get_subscriber_id(&self) -> i32;
fn get_cron_id(&self) -> Option<i32>;
fn from_input(input: Self::InputType, subscriber_id: i32) -> Self;
}
pub trait SystemTaskTrait: AsyncTaskTrait {}

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

@@ -12,10 +12,10 @@ pub use core::{
pub use config::TaskConfig;
pub use r#extern::{ApalisJobs, ApalisSchema};
pub use registry::{
OptimizeImageTask, SubscriberTask, SubscriberTaskType, SubscriberTaskTypeEnum,
SubscriberTaskTypeVariant, SubscriberTaskTypeVariantIter, SyncOneSubscriptionFeedsFullTask,
SyncOneSubscriptionFeedsIncrementalTask, SyncOneSubscriptionSourcesTask, SystemTask,
SystemTaskType, SystemTaskTypeEnum, SystemTaskTypeVariant, SystemTaskTypeVariantIter,
subscriber_task_schema,
OptimizeImageTask, SubscriberTask, SubscriberTaskInput, SubscriberTaskType,
SubscriberTaskTypeEnum, SubscriberTaskTypeVariant, SubscriberTaskTypeVariantIter,
SyncOneSubscriptionFeedsFullTask, SyncOneSubscriptionFeedsIncrementalTask,
SyncOneSubscriptionSourcesTask, SystemTask, SystemTaskType, SystemTaskTypeEnum,
SystemTaskTypeVariant, SystemTaskTypeVariantIter,
};
pub use service::TaskService;

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

@@ -2,10 +2,9 @@ mod subscriber;
mod system;
pub use subscriber::{
SubscriberTask, SubscriberTaskType, SubscriberTaskTypeEnum, SubscriberTaskTypeVariant,
SubscriberTaskTypeVariantIter, SyncOneSubscriptionFeedsFullTask,
SubscriberTask, SubscriberTaskInput, SubscriberTaskType, SubscriberTaskTypeEnum,
SubscriberTaskTypeVariant, SubscriberTaskTypeVariantIter, SyncOneSubscriptionFeedsFullTask,
SyncOneSubscriptionFeedsIncrementalTask, SyncOneSubscriptionSourcesTask,
subscriber_task_schema,
};
pub use system::{
OptimizeImageTask, SystemTask, SystemTaskType, SystemTaskTypeEnum, SystemTaskTypeVariant,

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

@@ -6,16 +6,36 @@ macro_rules! register_subscriber_task_type {
}
) => {
$(#[$type_meta])*
#[derive(typed_builder::TypedBuilder, schemars::JsonSchema, ts_rs::TS)]
#[derive(typed_builder::TypedBuilder, ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[ts(export, rename_all = "camelCase")]
$task_vis struct $task_name {
$($(#[$field_meta])* pub $field_name: $field_type,)*
pub subscriber_id: i32,
#[builder(default = None)]
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cron_id: Option<i32>,
}
paste::paste! {
$(#[$type_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
#[ts(export, 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::SubscriberTaskTrait for $task_name {
paste::paste! {
type InputType = [<$task_name Input>];
}
fn get_subscriber_id(&self) -> i32 {
self.subscriber_id
}
@@ -23,6 +43,14 @@ macro_rules! register_subscriber_task_type {
fn get_cron_id(&self) -> Option<i32> {
self.cron_id
}
fn from_input(input: Self::InputType, subscriber_id: i32) -> Self {
Self {
$($field_name: input.$field_name,)*
cron_id: input.cron_id,
subscriber_id: input.subscriber_id.unwrap_or(subscriber_id),
}
}
}
}
}

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

@@ -1,22 +1,17 @@
mod base;
mod subscription;
use jsonschema::Validator;
use once_cell::sync::OnceCell;
use schemars::JsonSchema;
use sea_orm::{DeriveActiveEnum, DeriveDisplay, EnumIter, FromJsonQueryResult};
use serde::{Deserialize, Serialize};
pub use subscription::{
SyncOneSubscriptionFeedsFullTask, SyncOneSubscriptionFeedsIncrementalTask,
SyncOneSubscriptionSourcesTask,
};
use ts_rs::TS;
macro_rules! register_subscriber_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
@@ -25,7 +20,7 @@ macro_rules! register_subscriber_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)
@@ -34,8 +29,9 @@ macro_rules! register_subscriber_task_types {
}
) => {
$(#[$type_enum_meta])*
#[derive(serde::Serialize, serde::Deserialize)]
#[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)]
@@ -46,9 +42,10 @@ macro_rules! register_subscriber_task_types {
$(#[$task_enum_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[serde(tag = "task_type")]
#[ts(export, rename_all = "camelCase", tag = "taskType")]
pub enum $task_enum_name {
#[ts(export,rename = "SubscriberTaskType", rename_all = "camelCase", tag = "taskType")]
$task_vis enum $task_enum_name {
$(
$(#[$task_variant_meta])*
#[serde(rename = $string_value)]
@@ -56,6 +53,20 @@ macro_rules! register_subscriber_task_types {
)*
}
paste::paste! {
$(#[$task_enum_meta])*
#[derive(ts_rs::TS, serde::Serialize, serde::Deserialize)]
#[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::SubscriberTaskTrait>::InputType),
)*
}
}
impl TryFrom<$task_enum_name> for serde_json::Value {
type Error = $crate::errors::RecorderError;
@@ -92,6 +103,10 @@ macro_rules! register_subscriber_task_types {
}
impl $crate::task::SubscriberTaskTrait for $task_enum_name {
paste::paste! {
type InputType = [<$task_enum_name Input>];
}
fn get_subscriber_id(&self) -> i32 {
match self {
$(Self::$task_variant(t) =>
@@ -105,6 +120,14 @@ macro_rules! register_subscriber_task_types {
<$task_type as $crate::task::SubscriberTaskTrait>::get_cron_id(t),)*
}
}
fn from_input(input: Self::InputType, subscriber_id: i32) -> Self {
match input {
$(Self::InputType::$task_variant(t) =>
Self::$task_variant(<$task_type as $crate::task::SubscriberTaskTrait>::from_input(t, subscriber_id)),)*
}
}
}
$(
@@ -122,8 +145,6 @@ register_subscriber_task_types!(
#[derive(
Clone,
Debug,
Serialize,
Deserialize,
PartialEq,
Eq,
Copy,
@@ -138,7 +159,7 @@ register_subscriber_task_types!(
}
},
task_enum: {
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, FromJsonQueryResult, JsonSchema, TS)]
#[derive(Clone, Debug, PartialEq, Eq, FromJsonQueryResult)]
pub enum SubscriberTask {
SyncOneSubscriptionFeedsIncremental(SyncOneSubscriptionFeedsIncrementalTask),
SyncOneSubscriptionFeedsFull(SyncOneSubscriptionFeedsFullTask),
@@ -146,15 +167,3 @@ register_subscriber_task_types!(
}
}
);
static SUBSCRIBER_TASK_SCHEMA: OnceCell<Validator> = OnceCell::new();
pub fn subscriber_task_schema() -> &'static Validator {
SUBSCRIBER_TASK_SCHEMA.get_or_init(|| {
let schema = schemars::schema_for!(SubscriberTask);
jsonschema::options()
.with_draft(jsonschema::Draft::Draft7)
.build(&serde_json::to_value(&schema).unwrap())
.unwrap()
})
}

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

@@ -1,5 +1,4 @@
use sea_orm::prelude::*;
use serde::{Deserialize, Serialize};
use super::base::register_subscriber_task_type;
use crate::{errors::RecorderResult, models::subscriptions::SubscriptionTrait};
@@ -40,7 +39,7 @@ macro_rules! register_subscription_task_type {
}
register_subscription_task_type! {
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SyncOneSubscriptionFeedsIncrementalTask {
} => async |subscription, ctx| -> RecorderResult<()> {
subscription.sync_feeds_incremental(ctx).await?;
@@ -49,7 +48,7 @@ register_subscription_task_type! {
}
register_subscription_task_type! {
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SyncOneSubscriptionFeedsFullTask {
} => async |subscription, ctx| -> RecorderResult<()> {
subscription.sync_feeds_full(ctx).await?;
@@ -58,7 +57,7 @@ register_subscription_task_type! {
}
register_subscription_task_type! {
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SyncOneSubscriptionSourcesTask {
} => async |subscription, ctx| -> RecorderResult<()> {
subscription.sync_sources(ctx).await?;