11296982085

Committed 11 Oct 2024 05:50PM UTC coverage: 55.96% (-0.03%) from 55.989%

Build # 11296982085

Build Type

push

github

Committed by

web-flow

Commit Message

Upgrade eslint-plugin-react-hooks (#1340)

Run Details

3934 of 7030 relevant lines covered (55.96%)

103.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.48

/src/routes/tasks.rs

use crate::{
    clients::aggregator_client::TaskUploadMetrics,
    config::FeatureFlags,
    entity::{Account, NewTask, Task, TaskColumn, Tasks, UpdateTask},
    Crypter, Db, Error, Permissions, PermissionsActor,
};
use querystrong::QueryStrong;
use sea_orm::{
    ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait, IntoActiveModel, ModelTrait,
    QueryFilter,
};
use std::time::Duration;
use time::OffsetDateTime;
use tokio::join;
use tracing::warn;
use trillium::{Conn, Handler, Status};
use trillium_api::{FromConn, Json, State};
use trillium_caching_headers::CachingHeadersExt;
use trillium_client::Client;
use trillium_router::RouterConnExt;

pub async fn index(_: &mut Conn, (account, db): (Account, Db)) -> Result<impl Handler, Error> {
    account
        .find_related(Tasks)
        .filter(TaskColumn::DeletedAt.is_null())
        .all(&db)
        .await
        .map(Json)
        .map_err(Error::from)
}

impl Permissions for Task {
    fn allow_write(&self, actor: &PermissionsActor) -> bool {
        actor.is_admin() || actor.account_ids().contains(&self.account_id)
    }
}

#[trillium::async_trait]
impl FromConn for Task {
    async fn from_conn(conn: &mut Conn) -> Option<Self> {
        let actor = PermissionsActor::from_conn(conn).await?;
        let db: &Db = conn.state()?;
        let id = conn.param("task_id")?;

        match Tasks::find_by_id(id).one(db).await {
            Ok(Some(task)) => actor.if_allowed(conn.method(), task),
            Ok(None) => None,
            Err(error) => {
                conn.insert_state(Error::from(error));
                None
            }
        }
    }
}

type CreateArgs = (Account, Json<NewTask>, State<Client>, Db);
pub async fn create(
    conn: &mut Conn,
    (account, mut task, State(client), db): CreateArgs,
) -> Result<impl Handler, Error> {
    let crypter = conn.state().unwrap();
    task.normalize_and_validate(account, &db)
        .await?
        .provision(client, crypter)
        .await?
        .insert(&db)
        .await
        .map_err(Into::into)
        .map(|task| (Status::Created, Json(task)))
}

async fn refresh_metrics_if_needed(
    task: Task,
    db: Db,
    client: Client,
    crypter: &Crypter,
) -> Result<Task, Error> {
    if OffsetDateTime::now_utc() - task.updated_at <= Duration::from_secs(5 * 60) {
        return Ok(task);
    }
    let aggregator = task.leader_aggregator(&db).await?;
    let metrics = if aggregator.features.upload_metrics_enabled() {
        aggregator
            .client(client, crypter)?
            .get_task_upload_metrics(&task.id)
            .await?
    } else {
        TaskUploadMetrics::default()
    };
    task.update_task_upload_metrics(metrics, db)
        .await
        .map_err(Into::into)
}

pub async fn show(
    conn: &mut Conn,
    (task, db, State(client), State(feature_flags)): (Task, Db, State<Client>, State<FeatureFlags>),
) -> Result<Json<Task>, Error> {
    let task = if feature_flags.metrics_refresh_enabled {
        let crypter = conn.state().unwrap();
        refresh_metrics_if_needed(task, db, client, crypter).await?
    } else {
        task
    };
    conn.set_last_modified(task.updated_at.into());
    Ok(Json(task))
}

type UpdateArgs = (Task, Json<UpdateTask>, State<Client>, Db);
pub async fn update(
    conn: &mut Conn,
    (task, Json(update), client, db): UpdateArgs,
) -> Result<impl Handler, Error> {
    let crypter = conn.state().unwrap();
    update
        .update(&client, &db, crypter, task)
        .await?
        .update(&db)
        .await
        .map(Json)
        .map_err(Error::from)
}

pub async fn delete(
    conn: &mut Conn,
    (task, client, db): (Task, State<Client>, Db),
) -> Result<impl Handler, Error> {
    let params = QueryStrong::parse(conn.querystring()).unwrap_or_default();
    let force = params
        .get_str("force")
        .and_then(|param| param.parse().ok())
        .unwrap_or(false);

    if task.deleted_at.is_some() {
        return Ok(Status::NoContent);
    }

    let crypter = conn.state().unwrap();
    let now = OffsetDateTime::now_utc();

    let mut am = task.clone().into_active_model();
    // If the task has not already expired, mark the aggregator-side tasks as expired. This will
    // allow the aggregators to cease processing and eventually GC the task at their leisure.
    if task.expiration.is_none() || task.expiration > Some(now) {
        let update = UpdateTask::expiration(Some(now));

        let (leader_result, helper_result) = join!(
            update.update_aggregator_expiration(
                task.leader_aggregator(&db).await?,
                &task.id,
                &client,
                crypter,
            ),
            update.update_aggregator_expiration(
                task.helper_aggregator(&db).await?,
                &task.id,
                &client,
                crypter,
            )
        );

        if force {
            let _ = leader_result
                .map_err(|err| warn!(?err, "failed to expire leader-side task, ignoring"));
            let _ = helper_result
                .map_err(|err| warn!(?err, "failed to expire helper-side task, ignoring"));
        } else {
            leader_result?;
            helper_result?;
        }

        am.expiration = ActiveValue::Set(Some(now));
    }

    am.updated_at = ActiveValue::Set(now);
    am.deleted_at = ActiveValue::Set(Some(now));
    am.update(&db).await?;
    Ok(Status::NoContent)
}

pub mod collector_auth_tokens {
    use super::*;
    pub async fn index(
        conn: &mut Conn,
        (task, db, State(client)): (Task, Db, State<Client>),
    ) -> Result<impl Handler, Error> {
        let leader = task.leader_aggregator(&db).await?;
        if leader.features.token_hash_enabled() {
            Err(Error::NotFound)
        } else {
            let crypter = conn.state().unwrap();
            let client = leader.client(client, crypter)?;
            let task_response = client.get_task(&task.id).await?;
            Ok(Json([task_response.collector_auth_token]))
        }
    }
}

1	use crate::{
2	clients::aggregator_client::TaskUploadMetrics,
3	config::FeatureFlags,
4	entity::{Account, NewTask, Task, TaskColumn, Tasks, UpdateTask},
5	Crypter, Db, Error, Permissions, PermissionsActor,
6	};
7	use querystrong::QueryStrong;
8	use sea_orm::{
9	ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait, IntoActiveModel, ModelTrait,
10	QueryFilter,
11	};
12	use std::time::Duration;
13	use time::OffsetDateTime;
14	use tokio::join;
15	use tracing::warn;
16	use trillium::{Conn, Handler, Status};
17	use trillium_api::{FromConn, Json, State};
18	use trillium_caching_headers::CachingHeadersExt;
19	use trillium_client::Client;
20	use trillium_router::RouterConnExt;
21
22	pub async fn index(_: &mut Conn, (account, db): (Account, Db)) -> Result<impl Handler, Error> {	10✔
23	account	10✔
24	.find_related(Tasks)	10✔
25	.filter(TaskColumn::DeletedAt.is_null())	10✔
26	.all(&db)	10✔
27	.await	28✔
28	.map(Json)	10✔
29	.map_err(Error::from)	10✔
30	}	10✔
31
32	impl Permissions for Task {
33	fn allow_write(&self, actor: &PermissionsActor) -> bool {	44✔
34	actor.is_admin() \|\| actor.account_ids().contains(&self.account_id)	44✔
35	}	44✔
36	}
37
38	#[trillium::async_trait]
39	impl FromConn for Task {
40	async fn from_conn(conn: &mut Conn) -> Option<Self> {	47✔
41	let actor = PermissionsActor::from_conn(conn).await?;	47✔
42	let db: &Db = conn.state()?;	47✔
43	let id = conn.param("task_id")?;	47✔
44
45	match Tasks::find_by_id(id).one(db).await {	97✔
46	Ok(Some(task)) => actor.if_allowed(conn.method(), task),	44✔
47	Ok(None) => None,	3✔
48	Err(error) => {	×
49	conn.insert_state(Error::from(error));	×
50	None	×
51	}
52	}
53	}	94✔
54	}
55
56	type CreateArgs = (Account, Json<NewTask>, State<Client>, Db);
57	pub async fn create(	10✔
58	conn: &mut Conn,	10✔
59	(account, mut task, State(client), db): CreateArgs,	10✔
60	) -> Result<impl Handler, Error> {	10✔
61	let crypter = conn.state().unwrap();	10✔
62	task.normalize_and_validate(account, &db)	10✔
63	.await?	60✔
64	.provision(client, crypter)	7✔
65	.await?	14✔
66	.insert(&db)	7✔
67	.await	35✔
68	.map_err(Into::into)	7✔
69	.map(\|task\| (Status::Created, Json(task)))	7✔
70	}	10✔
71
72	async fn refresh_metrics_if_needed(	8✔
73	task: Task,	8✔
74	db: Db,	8✔
75	client: Client,	8✔
76	crypter: &Crypter,	8✔
77	) -> Result<Task, Error> {	8✔
78	if OffsetDateTime::now_utc() - task.updated_at <= Duration::from_secs(5 * 60) {	8✔
79	return Ok(task);	6✔
80	}	2✔
81	let aggregator = task.leader_aggregator(&db).await?;	4✔
82	let metrics = if aggregator.features.upload_metrics_enabled() {	2✔
83	aggregator	1✔
84	.client(client, crypter)?	1✔
85	.get_task_upload_metrics(&task.id)	1✔
86	.await?	×
87	} else {
88	TaskUploadMetrics::default()	1✔
89	};
90	task.update_task_upload_metrics(metrics, db)	2✔
91	.await	8✔
92	.map_err(Into::into)	2✔
93	}	8✔
94
95	pub async fn show(	8✔
96	conn: &mut Conn,	8✔
97	(task, db, State(client), State(feature_flags)): (Task, Db, State<Client>, State<FeatureFlags>),	8✔
98	) -> Result<Json<Task>, Error> {	8✔
99	let task = if feature_flags.metrics_refresh_enabled {	8✔
100	let crypter = conn.state().unwrap();	8✔
101	refresh_metrics_if_needed(task, db, client, crypter).await?	12✔
102	} else {
103	task	×
104	};
105	conn.set_last_modified(task.updated_at.into());	8✔
106	Ok(Json(task))	8✔
107	}	8✔
108
109	type UpdateArgs = (Task, Json<UpdateTask>, State<Client>, Db);
110	pub async fn update(	11✔
111	conn: &mut Conn,	11✔
112	(task, Json(update), client, db): UpdateArgs,	11✔
113	) -> Result<impl Handler, Error> {	11✔
114	let crypter = conn.state().unwrap();	11✔
115	update	11✔
116	.update(&client, &db, crypter, task)	11✔
117	.await?	21✔
118	.update(&db)	10✔
119	.await	34✔
120	.map(Json)	10✔
121	.map_err(Error::from)	10✔
122	}	11✔
123
124	pub async fn delete(	10✔
125	conn: &mut Conn,	10✔
126	(task, client, db): (Task, State<Client>, Db),	10✔
127	) -> Result<impl Handler, Error> {	10✔
128	let params = QueryStrong::parse(conn.querystring()).unwrap_or_default();	10✔
129	let force = params	10✔
130	.get_str("force")	10✔
131	.and_then(\|param\| param.parse().ok())	10✔
132	.unwrap_or(false);	10✔
133		10✔
134	if task.deleted_at.is_some() {	10✔
135	return Ok(Status::NoContent);	1✔
136	}	9✔
137		9✔
138	let crypter = conn.state().unwrap();	9✔
139	let now = OffsetDateTime::now_utc();	9✔
140		9✔
141	let mut am = task.clone().into_active_model();	9✔
142	// If the task has not already expired, mark the aggregator-side tasks as expired. This will	9✔
143	// allow the aggregators to cease processing and eventually GC the task at their leisure.	9✔
144	if task.expiration.is_none() \|\| task.expiration > Some(now) {	9✔
145	let update = UpdateTask::expiration(Some(now));	8✔
146
147	let (leader_result, helper_result) = join!(	8✔
148	update.update_aggregator_expiration(	8✔
149	task.leader_aggregator(&db).await?,	17✔
150	&task.id,	8✔
151	&client,	8✔
152	crypter,	8✔
153	),	8✔
154	update.update_aggregator_expiration(	8✔
155	task.helper_aggregator(&db).await?,	19✔
156	&task.id,	8✔
157	&client,	8✔
158	crypter,	8✔
159	)
160	);
161
162	if force {	8✔
163	let _ = leader_result	2✔
164	.map_err(\|err\| warn!(?err, "failed to expire leader-side task, ignoring"));	2✔
165	let _ = helper_result	2✔
166	.map_err(\|err\| warn!(?err, "failed to expire helper-side task, ignoring"));	2✔
167	} else {	2✔
168	leader_result?;	6✔
169	helper_result?;	6✔
170	}
171
172	am.expiration = ActiveValue::Set(Some(now));	8✔
173	}	1✔
174
175	am.updated_at = ActiveValue::Set(now);	9✔
176	am.deleted_at = ActiveValue::Set(Some(now));	9✔
177	am.update(&db).await?;	39✔
178	Ok(Status::NoContent)	9✔
179	}	10✔
180
181	pub mod collector_auth_tokens {
182	use super::*;
183	pub async fn index(	7✔
184	conn: &mut Conn,	7✔
185	(task, db, State(client)): (Task, Db, State<Client>),	7✔
186	) -> Result<impl Handler, Error> {	7✔
187	let leader = task.leader_aggregator(&db).await?;	14✔
188	if leader.features.token_hash_enabled() {	7✔
189	Err(Error::NotFound)	2✔
190	} else {
191	let crypter = conn.state().unwrap();	5✔
192	let client = leader.client(client, crypter)?;	5✔
193	let task_response = client.get_task(&task.id).await?;	5✔
194	Ok(Json([task_response.collector_auth_token]))	5✔
195	}
196	}	7✔
197	}

divviup / divviup-api / 11296982085

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous