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

fix: fix graphql

Browse Source
This commit is contained in:
master
2025-05-22 02:11:16 +08:00
parent b2f327d48f
commit f1d8318500
15 changed files with 740 additions and 447 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
apps/recorder/Cargo.toml
View File

@@ -25,6 +25,11 @@ testcontainers = [
]
[dependencies]
downloader = { workspace = true }
util = { workspace = true }
util-derive = { workspace = true }
fetch = { workspace = true }
serde = { workspace = true }
tokio = { workspace = true }
serde_json = { workspace = true }
@@ -49,7 +54,7 @@ serde_with = { workspace = true }
moka = { workspace = true }
chrono = { workspace = true }
tracing-subscriber = { workspace = true }
mockito = { workspace = true, optional = true }
sea-orm = { version = "1.1", features = [
"sqlx-sqlite",
@@ -113,15 +118,12 @@ cocoon = { version = "0.4.3", features = ["getrandom", "thiserror"] }
rand = "0.9.1"
rust_decimal = "1.37.1"
reqwest_cookie_store = "0.8.0"
mockito = { version = "1.6.1", optional = true }
downloader = { workspace = true }
util = { workspace = true }
fetch = { workspace = true }
nanoid = "0.4.0"
[dev-dependencies]
serial_test = "3"
insta = { version = "1", features = ["redactions", "yaml", "filters"] }
rstest = "0.25"
ctor = "0.4.0"
mockito = { workspace = true }

556
apps/recorder/src/graphql/infra/filter/json.rs
View File

File diff suppressed because it is too large Load Diff

12
apps/recorder/src/graphql/infra/filter/mod.rs
View File

@@ -1,10 +1,11 @@
mod json;
mod subscriber;
use std::borrow::Cow;
use async_graphql::dynamic::TypeRef;
pub use json::{JSONB_FILTER_INFO, jsonb_filter_condition_function};
pub use json::{
JSONB_FILTER_NAME, jsonb_filter_condition_function,
register_jsonb_input_filter_to_dynamic_schema,
};
use maplit::btreeset;
use seaography::{FilterInfo, FilterOperation as SeaographqlFilterOperation};
pub use subscriber::{SUBSCRIBER_ID_FILTER_INFO, subscriber_id_condition_function};
@@ -15,9 +16,4 @@ pub fn init_custom_filter_info() {
base_type: TypeRef::INT.into(),
supported_operations: btreeset! { SeaographqlFilterOperation::Equals },
});
JSONB_FILTER_INFO.get_or_init(|| FilterInfo {
type_name: String::from("JsonbFilterInput"),
base_type: TypeRef::Named(Cow::Borrowed("serde_json::Value")).to_string(),
supported_operations: btreeset! { SeaographqlFilterOperation::Equals },
});
}

38
apps/recorder/src/graphql/schema_root.rs
View File

@@ -3,14 +3,17 @@ use once_cell::sync::OnceCell;
use sea_orm::{DatabaseConnection, EntityTrait, Iterable};
use seaography::{Builder, BuilderContext, FilterType, FilterTypesMapHelper};
use crate::graphql::infra::{
filter::{
JSONB_FILTER_INFO, SUBSCRIBER_ID_FILTER_INFO, init_custom_filter_info,
subscriber_id_condition_function,
use crate::graphql::{
infra::{
filter::{
JSONB_FILTER_NAME, SUBSCRIBER_ID_FILTER_INFO, init_custom_filter_info,
register_jsonb_input_filter_to_dynamic_schema, subscriber_id_condition_function,
},
guard::{guard_entity_with_subscriber_id, guard_field_with_subscriber_id},
transformer::{filter_condition_transformer, mutation_input_object_transformer},
util::{get_entity_column_key, get_entity_key},
},
guard::{guard_entity_with_subscriber_id, guard_field_with_subscriber_id},
transformer::{filter_condition_transformer, mutation_input_object_transformer},
util::{get_entity_column_key, get_entity_key},
views::register_subscriptions_to_schema,
};
pub static CONTEXT: OnceCell<BuilderContext> = OnceCell::new();
@@ -35,9 +38,7 @@ where
let entity_column_key = get_entity_column_key::<T>(context, column);
context.filter_types.overwrites.insert(
entity_column_key.clone(),
Some(FilterType::Custom(
JSONB_FILTER_INFO.get().unwrap().type_name.clone(),
)),
Some(FilterType::Custom(JSONB_FILTER_NAME.to_string())),
);
}
@@ -94,12 +95,12 @@ pub fn schema(
let context = CONTEXT.get_or_init(|| {
let mut context = BuilderContext::default();
context.pagination_input.type_name = "SeaographyPaginationInput".to_string();
context.pagination_info_object.type_name = "SeaographyPaginationInfo".to_string();
context.cursor_input.type_name = "SeaographyCursorInput".to_string();
context.offset_input.type_name = "SeaographyOffsetInput".to_string();
context.page_input.type_name = "SeaographyPageInput".to_string();
context.page_info_object.type_name = "SeaographyPageInfo".to_string();
context.pagination_input.type_name = "PaginationInput".to_string();
context.pagination_info_object.type_name = "PaginationInfo".to_string();
context.cursor_input.type_name = "CursorInput".to_string();
context.offset_input.type_name = "OffsetInput".to_string();
context.page_input.type_name = "PageInput".to_string();
context.page_info_object.type_name = "PageInfo".to_string();
restrict_subscriber_for_entity::<bangumi::Entity>(
&mut context,
@@ -160,6 +161,7 @@ pub fn schema(
builder.schema = builder.schema.register(
filter_types_map_helper.generate_filter_input(SUBSCRIBER_ID_FILTER_INFO.get().unwrap()),
);
builder.schema = register_jsonb_input_filter_to_dynamic_schema(builder.schema);
}
{
@@ -193,6 +195,10 @@ pub fn schema(
builder.register_enumeration::<downloads::DownloadMime>();
}
{
builder = register_subscriptions_to_schema(builder);
}
let schema = builder.schema_builder();
let schema = if let Some(depth) = depth {

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

@@ -1,2 +1,3 @@
mod subscription;
mod task;
pub use subscription::register_subscriptions_to_schema;

308
apps/recorder/src/graphql/views/subscription.rs
View File

@@ -1,6 +1,11 @@
use std::sync::Arc;
use async_graphql::{Context, InputObject, Object, Result as GraphQLResult, SimpleObject};
use async_graphql::dynamic::{
Field, FieldFuture, FieldValue, InputObject, InputValue, Object, TypeRef,
};
use seaography::Builder as SeaographyBuilder;
use serde::{Deserialize, Serialize};
use util_derive::DynamicGraphql;
use crate::{
app::AppContextTrait,
@@ -9,116 +14,213 @@ use crate::{
task::SubscriberTaskPayload,
};
pub struct SubscriptionMutation;
#[derive(InputObject)]
#[derive(DynamicGraphql, Serialize, Deserialize, Clone, Debug)]
struct SyncOneSubscriptionFilterInput {
pub subscription_id: i32,
}
#[derive(SimpleObject)]
struct SyncOneSubscriptionTaskOutput {
impl SyncOneSubscriptionFilterInput {
fn input_type_name() -> &'static str {
"SyncOneSubscriptionFilterInput"
}
fn arg_name() -> &'static str {
"filter"
}
fn generate_input_object() -> InputObject {
InputObject::new(Self::input_type_name())
.description("The input of the subscriptionSyncOne series of mutations")
.field(InputValue::new(
SyncOneSubscriptionFilterInputFieldEnum::SubscriptionId.as_str(),
TypeRef::named_nn(TypeRef::INT),
))
}
}
#[derive(DynamicGraphql, Serialize, Deserialize, Clone, Debug)]
pub struct SyncOneSubscriptionInfo {
pub task_id: String,
}
#[Object]
impl SubscriptionMutation {
async fn sync_one_subscription_feeds_incremental(
&self,
ctx: &Context<'_>,
input: SyncOneSubscriptionFilterInput,
) -> GraphQLResult<SyncOneSubscriptionTaskOutput> {
let auth_user_info = ctx.data::<AuthUserInfo>()?;
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let subscriber_id = auth_user_info.subscriber_auth.subscriber_id;
let subscription_model = subscriptions::Model::find_by_id_and_subscriber_id(
app_ctx.as_ref(),
input.subscription_id,
subscriber_id,
)
.await?;
let subscription = subscriptions::Subscription::try_from_model(&subscription_model)?;
let task_service = app_ctx.task();
let task_id = task_service
.add_subscriber_task(
auth_user_info.subscriber_auth.subscriber_id,
SubscriberTaskPayload::SyncOneSubscriptionFeedsIncremental(subscription.into()),
)
.await?;
Ok(SyncOneSubscriptionTaskOutput {
task_id: task_id.to_string(),
})
impl SyncOneSubscriptionInfo {
fn object_type_name() -> &'static str {
"SyncOneSubscriptionInfo"
}
async fn sync_one_subscription_feeds_full(
&self,
ctx: &Context<'_>,
input: SyncOneSubscriptionFilterInput,
) -> GraphQLResult<SyncOneSubscriptionTaskOutput> {
let auth_user_info = ctx.data::<AuthUserInfo>()?;
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let subscriber_id = auth_user_info.subscriber_auth.subscriber_id;
let subscription_model = subscriptions::Model::find_by_id_and_subscriber_id(
app_ctx.as_ref(),
input.subscription_id,
subscriber_id,
)
.await?;
let subscription = subscriptions::Subscription::try_from_model(&subscription_model)?;
let task_service = app_ctx.task();
let task_id = task_service
.add_subscriber_task(
auth_user_info.subscriber_auth.subscriber_id,
SubscriberTaskPayload::SyncOneSubscriptionFeedsFull(subscription.into()),
)
.await?;
Ok(SyncOneSubscriptionTaskOutput {
task_id: task_id.to_string(),
})
}
async fn sync_one_subscription_sources(
&self,
ctx: &Context<'_>,
input: SyncOneSubscriptionFilterInput,
) -> GraphQLResult<SyncOneSubscriptionTaskOutput> {
let auth_user_info = ctx.data::<AuthUserInfo>()?;
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let subscriber_id = auth_user_info.subscriber_auth.subscriber_id;
let subscription_model = subscriptions::Model::find_by_id_and_subscriber_id(
app_ctx.as_ref(),
input.subscription_id,
subscriber_id,
)
.await?;
let subscription = subscriptions::Subscription::try_from_model(&subscription_model)?;
let task_service = app_ctx.task();
let task_id = task_service
.add_subscriber_task(
auth_user_info.subscriber_auth.subscriber_id,
SubscriberTaskPayload::SyncOneSubscriptionSources(subscription.into()),
)
.await?;
Ok(SyncOneSubscriptionTaskOutput {
task_id: task_id.to_string(),
})
fn generate_output_object() -> Object {
Object::new(Self::object_type_name())
.description("The output of the subscriptionSyncOne series of mutations")
.field(Field::new(
"taskId",
TypeRef::named_nn(TypeRef::STRING),
move |ctx| {
FieldFuture::new(async move {
let subscription_info = ctx.parent_value.try_downcast_ref::<Self>()?;
Ok(Some(async_graphql::Value::from(
subscription_info.task_id.as_str(),
)))
})
},
))
}
}
pub fn register_subscriptions_to_schema(mut builder: SeaographyBuilder) -> SeaographyBuilder {
builder.schema = builder
.schema
.register(SyncOneSubscriptionFilterInput::generate_input_object());
builder.schema = builder
.schema
.register(SyncOneSubscriptionInfo::generate_output_object());
builder.queries.push(
Field::new(
"subscriptionSyncOneFeedsIncremental",
TypeRef::named_nn(SyncOneSubscriptionInfo::object_type_name()),
move |ctx| {
FieldFuture::new(async move {
let auth_user_info = ctx.data::<AuthUserInfo>()?;
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let subscriber_id = auth_user_info.subscriber_auth.subscriber_id;
let filter_input: SyncOneSubscriptionFilterInput = ctx
.args
.get(SyncOneSubscriptionFilterInput::arg_name())
.unwrap()
.deserialize()?;
let subscription_model = subscriptions::Model::find_by_id_and_subscriber_id(
app_ctx.as_ref(),
filter_input.subscription_id,
subscriber_id,
)
.await?;
let subscription =
subscriptions::Subscription::try_from_model(&subscription_model)?;
let task_service = app_ctx.task();
let task_id = task_service
.add_subscriber_task(
auth_user_info.subscriber_auth.subscriber_id,
SubscriberTaskPayload::SyncOneSubscriptionFeedsIncremental(
subscription.into(),
),
)
.await?;
Ok(Some(FieldValue::owned_any(SyncOneSubscriptionInfo {
task_id: task_id.to_string(),
})))
})
},
)
.argument(InputValue::new(
SyncOneSubscriptionFilterInput::arg_name(),
TypeRef::named_nn(SyncOneSubscriptionFilterInput::input_type_name()),
)),
);
builder.queries.push(
Field::new(
"subscriptionSyncOneFeedsFull",
TypeRef::named_nn(SyncOneSubscriptionInfo::object_type_name()),
move |ctx| {
FieldFuture::new(async move {
let auth_user_info = ctx.data::<AuthUserInfo>()?;
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let subscriber_id = auth_user_info.subscriber_auth.subscriber_id;
let filter_input: SyncOneSubscriptionFilterInput = ctx
.args
.get(SyncOneSubscriptionFilterInput::arg_name())
.unwrap()
.deserialize()?;
let subscription_model = subscriptions::Model::find_by_id_and_subscriber_id(
app_ctx.as_ref(),
filter_input.subscription_id,
subscriber_id,
)
.await?;
let subscription =
subscriptions::Subscription::try_from_model(&subscription_model)?;
let task_service = app_ctx.task();
let task_id = task_service
.add_subscriber_task(
auth_user_info.subscriber_auth.subscriber_id,
SubscriberTaskPayload::SyncOneSubscriptionFeedsFull(
subscription.into(),
),
)
.await?;
Ok(Some(FieldValue::owned_any(SyncOneSubscriptionInfo {
task_id: task_id.to_string(),
})))
})
},
)
.argument(InputValue::new(
SyncOneSubscriptionFilterInput::arg_name(),
TypeRef::named_nn(SyncOneSubscriptionFilterInput::input_type_name()),
)),
);
builder.mutations.push(
Field::new(
"subscriptionSyncOneSources",
TypeRef::named_nn(SyncOneSubscriptionInfo::object_type_name()),
move |ctx| {
FieldFuture::new(async move {
let auth_user_info = ctx.data::<AuthUserInfo>()?;
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let subscriber_id = auth_user_info.subscriber_auth.subscriber_id;
let filter_input: SyncOneSubscriptionFilterInput = ctx
.args
.get(SyncOneSubscriptionFilterInput::arg_name())
.unwrap()
.deserialize()?;
let subscription_model = subscriptions::Model::find_by_id_and_subscriber_id(
app_ctx.as_ref(),
filter_input.subscription_id,
subscriber_id,
)
.await?;
let subscription =
subscriptions::Subscription::try_from_model(&subscription_model)?;
let task_service = app_ctx.task();
let task_id = task_service
.add_subscriber_task(
auth_user_info.subscriber_auth.subscriber_id,
SubscriberTaskPayload::SyncOneSubscriptionSources(subscription.into()),
)
.await?;
Ok(Some(FieldValue::owned_any(SyncOneSubscriptionInfo {
task_id: task_id.to_string(),
})))
})
},
)
.argument(InputValue::new(
SyncOneSubscriptionFilterInput::arg_name(),
TypeRef::named_nn(SyncOneSubscriptionFilterInput::input_type_name()),
)),
);
builder
}

27
apps/recorder/src/graphql/views/task.rs
View File

@@ -1,27 +0,0 @@
use std::sync::Arc;
use async_graphql::{Context, InputObject, Object, Result as GraphQLResult};
use crate::{app::AppContextTrait, auth::AuthUserInfo};
struct TaskQuery;
#[derive(InputObject)]
struct SubscriberTasksFilterInput {
pub subscription_id: Option<i32>,
pub task_id: Option<String>,
pub task_type: Option<String>,
}
#[Object]
impl TaskQuery {
async fn subscriber_tasks(&self, ctx: &Context<'_>) -> GraphQLResult<Vec<String>> {
let auth_user_info = ctx.data::<AuthUserInfo>()?;
let app_ctx = ctx.data::<Arc<dyn AppContextTrait>>()?;
let subscriber_id = auth_user_info.subscriber_auth.subscriber_id;
let task_service = app_ctx.task();
todo!()
}
}

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

@@ -12,10 +12,10 @@ impl MigrationTrait for Migration {
let db = manager.get_connection();
db.execute_unprepared(&format!(
r#"CREATE VIEW IF NOT EXISTS subscriber_task AS
r#"CREATE OR REPLACE VIEW subscriber_task AS
SELECT
job,
task_type,
job_type,
status,
(job->'subscriber_id')::integer AS subscriber_id,
(job->'task_type')::text AS task_type,
@@ -29,7 +29,7 @@ SELECT
done_at,
priority
FROM apalis.jobs
WHERE job_type = {SUBSCRIBER_TASK_APALIS_NAME}
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")')"#,
))
@@ -38,7 +38,7 @@ AND jsonb_path_exists(job, '$.task_type ? (@.type() == "string")')"#,
db.execute_unprepared(&format!(
r#"CREATE INDEX IF NOT EXISTS idx_apalis_jobs_subscriber_id
ON apalis.jobs ((job -> 'subscriber_id'))
WHERE job_type = {SUBSCRIBER_TASK_APALIS_NAME}
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")')"#
))

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

@@ -1,6 +1,5 @@
use std::sync::Arc;
use apalis::prelude::State;
use futures::Stream;
use serde::{Serialize, de::DeserializeOwned};

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

@@ -55,7 +55,9 @@ pub async fn build_testing_database_service(
}
#[cfg(not(feature = "testcontainers"))]
pub async fn build_testing_database_service() -> RecorderResult<DatabaseService> {
pub async fn build_testing_database_service(
config: TestingDatabaseServiceConfig,
) -> RecorderResult<DatabaseService> {
let db_service = DatabaseService::from_config(DatabaseConfig {
uri: String::from("postgres://konobangu:konobangu@127.0.0.1:5432/konobangu"),
enable_logging: true,
@@ -64,7 +66,7 @@ pub async fn build_testing_database_service() -> RecorderResult<DatabaseService>
connect_timeout: 5000,
idle_timeout: 10000,
acquire_timeout: None,
auto_migrate: true,
auto_migrate: config.auto_migrate,
})
.await?;