1mod operations;
2pub mod test_utils;
3
4use std::cmp::max;
5use std::collections::BTreeMap;
6use std::path::Path;
7
8use chrono::{DateTime, Utc};
9use kellnr_common::crate_data::{CrateData, CrateRegistryDep, CrateVersionData};
10use kellnr_common::crate_overview::CrateOverview;
11use kellnr_common::cratesio_prefetch_msg::{CratesioPrefetchMsg, UpdateData};
12use kellnr_common::index_metadata::{IndexDep, IndexMetadata};
13use kellnr_common::normalized_name::NormalizedName;
14use kellnr_common::original_name::OriginalName;
15use kellnr_common::prefetch::Prefetch;
16use kellnr_common::publish_metadata::PublishMetadata;
17use kellnr_common::version::Version;
18use kellnr_common::webhook::{Webhook, WebhookEvent, WebhookQueue};
19use kellnr_entity::prelude::*;
20use kellnr_entity::{
21 auth_token, crate_author, crate_author_to_crate, crate_category, crate_category_to_crate,
22 crate_group, crate_index, crate_keyword, crate_keyword_to_crate, crate_meta, crate_user,
23 cratesio_crate, cratesio_index, cratesio_meta, doc_queue, group, group_user, krate,
24 oauth2_identity, oauth2_state, owner, session, toolchain, toolchain_component,
25 toolchain_target, user, webhook, webhook_queue,
26};
27use kellnr_migration::iden::{
28 AuthTokenIden, CrateIden, CrateMetaIden, CratesIoIden, CratesIoMetaIden, GroupIden,
29};
30use sea_orm::entity::prelude::Uuid;
31use sea_orm::prelude::async_trait::async_trait;
32use sea_orm::query::{QueryOrder, QuerySelect, TransactionTrait};
33use sea_orm::sea_query::{Alias, Cond, Expr, Iden, JoinType, Order, Query, UnionType};
34use sea_orm::{
35 ActiveModelTrait, ActiveValue, ColumnTrait, DatabaseConnection, EntityTrait, ExprTrait,
36 FromQueryResult, ModelTrait, QueryFilter, RelationTrait, Set,
37};
38
39use crate::error::DbError;
40use crate::password::{generate_salt, hash_pwd, hash_token};
41use crate::provider::{
42 ChannelInfo, DbResult, OAuth2StateData, PrefetchState, SessionInfo, ToolchainComponentInfo,
43 ToolchainTargetInfo, ToolchainWithTargets,
44};
45use crate::tables::init_database;
46use crate::{
47 AuthToken, ConString, CrateMeta, CrateSummary, DbProvider, DocQueueEntry, Group, User,
48};
49
50const DB_DATE_FORMAT: &str = "%Y-%m-%d %H:%M:%S";
51
52const MAX_SESSION_AGE: std::time::Duration =
56 std::time::Duration::from_secs(100 * 365 * 24 * 60 * 60);
57
58fn session_age_to_duration(d: std::time::Duration) -> Option<chrono::Duration> {
63 if d.is_zero() {
64 return None;
65 }
66 let clamped = d.min(MAX_SESSION_AGE);
67 Some(chrono::Duration::from_std(clamped).expect("clamped session age fits chrono::Duration"))
70}
71
72pub struct Database {
73 db_con: DatabaseConnection,
74 session_age: Option<chrono::Duration>,
77}
78
79impl Database {
80 pub async fn new(con: &ConString, max_con: u32) -> Result<Self, DbError> {
81 let db_con = init_database(con, max_con)
82 .await
83 .map_err(|e| DbError::InitializationError(e.to_string()))?;
84
85 if operations::no_user_exists(&db_con).await? {
86 operations::insert_admin_credentials(&db_con, con).await?;
87 }
88
89 Ok(Self {
90 db_con,
91 session_age: session_age_to_duration(con.session_age()),
92 })
93 }
94
95 fn session_expiry_cutoff(&self) -> Option<String> {
99 self.session_age
100 .map(|age| (Utc::now() - age).format(DB_DATE_FORMAT).to_string())
101 }
102
103 async fn get_user_model(&self, name: &str) -> DbResult<user::Model> {
105 user::Entity::find()
106 .filter(user::Column::Name.eq(name))
107 .one(&self.db_con)
108 .await?
109 .ok_or_else(|| DbError::UserNotFound(name.to_owned()))
110 }
111
112 async fn get_krate_model(&self, crate_name: &NormalizedName) -> DbResult<krate::Model> {
114 krate::Entity::find()
115 .filter(krate::Column::Name.eq(crate_name))
116 .one(&self.db_con)
117 .await?
118 .ok_or_else(|| DbError::CrateNotFound(crate_name.to_string()))
119 }
120
121 async fn get_group_model(&self, name: &str) -> DbResult<group::Model> {
123 group::Entity::find()
124 .filter(group::Column::Name.eq(name))
125 .one(&self.db_con)
126 .await?
127 .ok_or_else(|| DbError::GroupNotFound(name.to_owned()))
128 }
129
130 async fn toolchain_with_targets(&self, tc: toolchain::Model) -> DbResult<ToolchainWithTargets> {
132 let targets = toolchain_target::Entity::find()
133 .filter(toolchain_target::Column::ToolchainFk.eq(tc.id))
134 .all(&self.db_con)
135 .await?;
136
137 let mut target_infos = Vec::with_capacity(targets.len());
138 for t in targets {
139 let components = toolchain_component::Entity::find()
140 .filter(toolchain_component::Column::ToolchainTargetFk.eq(t.id))
141 .all(&self.db_con)
142 .await?;
143 target_infos.push(ToolchainTargetInfo {
144 id: t.id,
145 target: t.target,
146 storage_path: t.storage_path,
147 hash: t.hash,
148 size: t.size,
149 status: t.status,
150 components: components
151 .into_iter()
152 .map(ToolchainComponentInfo::from)
153 .collect(),
154 });
155 }
156
157 Ok(ToolchainWithTargets {
158 id: tc.id,
159 name: tc.name,
160 version: tc.version,
161 date: tc.date,
162 channel: tc.channel,
163 created: tc.created,
164 targets: target_infos,
165 })
166 }
167
168 async fn get_crate_index_model(
170 &self,
171 crate_name: &NormalizedName,
172 version: &Version,
173 ) -> DbResult<crate_index::Model> {
174 crate_index::Entity::find()
175 .filter(crate_index::Column::Name.eq(crate_name))
176 .filter(crate_index::Column::Vers.eq(version))
177 .one(&self.db_con)
178 .await?
179 .ok_or_else(|| DbError::CrateIndexNotFound(crate_name.to_string(), version.to_string()))
180 }
181
182 async fn get_toolchain_by_name_version(
184 &self,
185 name: &str,
186 version: &str,
187 ) -> DbResult<Option<toolchain::Model>> {
188 toolchain::Entity::find()
189 .filter(toolchain::Column::Name.eq(name))
190 .filter(toolchain::Column::Version.eq(version))
191 .one(&self.db_con)
192 .await
193 .map_err(Into::into)
194 }
195
196 async fn get_crate_group_by_name_and_group(
198 &self,
199 crate_name: &NormalizedName,
200 group: &str,
201 ) -> DbResult<Option<crate_group::Model>> {
202 crate_group::Entity::find()
203 .join(JoinType::InnerJoin, crate_group::Relation::Krate.def())
204 .join(JoinType::InnerJoin, crate_group::Relation::Group.def())
205 .filter(
206 Cond::all()
207 .add(krate::Column::Name.eq(crate_name))
208 .add(group::Column::Name.eq(group)),
209 )
210 .one(&self.db_con)
211 .await
212 .map_err(Into::into)
213 }
214
215 async fn get_group_user_by_group_and_user(
217 &self,
218 group_name: &str,
219 user: &str,
220 ) -> DbResult<Option<group_user::Model>> {
221 group_user::Entity::find()
222 .join(JoinType::InnerJoin, group_user::Relation::Group.def())
223 .join(JoinType::InnerJoin, group_user::Relation::User.def())
224 .filter(
225 Cond::all()
226 .add(group::Column::Name.eq(group_name))
227 .add(user::Column::Name.eq(user)),
228 )
229 .one(&self.db_con)
230 .await
231 .map_err(Into::into)
232 }
233
234 async fn get_crate_user_by_crate_and_user(
236 &self,
237 crate_name: &NormalizedName,
238 user: &str,
239 ) -> DbResult<Option<crate_user::Model>> {
240 crate_user::Entity::find()
241 .join(JoinType::InnerJoin, crate_user::Relation::Krate.def())
242 .join(JoinType::InnerJoin, crate_user::Relation::User.def())
243 .filter(
244 Cond::all()
245 .add(krate::Column::Name.eq(crate_name))
246 .add(user::Column::Name.eq(user)),
247 )
248 .one(&self.db_con)
249 .await
250 .map_err(Into::into)
251 }
252
253 async fn query_crates(
255 &self,
256 contains: Option<&str>,
257 limit_offset: Option<(u64, u64)>,
258 cache: bool,
259 ) -> DbResult<Vec<CrateOverview>> {
260 let mut query = Query::select();
261 query
262 .expr_as(Expr::col(CrateIden::OriginalName), Alias::new("name"))
263 .expr_as(Expr::col(CrateIden::MaxVersion), Alias::new("version"))
264 .expr_as(Expr::col(CrateIden::LastUpdated), Alias::new("date"))
265 .expr_as(
266 Expr::col(CrateIden::TotalDownloads),
267 Alias::new("total_downloads"),
268 )
269 .expr_as(Expr::col(CrateIden::Description), Alias::new("description"))
270 .expr_as(
271 Expr::col(CrateMetaIden::Documentation),
272 Alias::new("documentation"),
273 )
274 .expr_as(Expr::cust("false"), Alias::new("is_cache"))
275 .from(CrateMetaIden::Table)
276 .inner_join(
277 CrateIden::Table,
278 Expr::col((CrateMetaIden::Table, CrateMetaIden::CrateFk))
279 .equals((CrateIden::Table, CrateIden::Id)),
280 )
281 .and_where(
282 Expr::col((CrateMetaIden::Table, CrateMetaIden::Version))
283 .equals((CrateIden::Table, CrateIden::MaxVersion)),
284 );
285
286 if let Some(contains) = contains {
288 query.and_where(
289 Expr::col((CrateIden::Table, CrateIden::Name)).like(format!("%{contains}%")),
290 );
291 }
292
293 if cache {
295 let mut query2 = Query::select();
296 query2
297 .expr_as(Expr::col(CratesIoIden::OriginalName), Alias::new("name"))
298 .expr_as(Expr::col(CratesIoIden::MaxVersion), Alias::new("version"))
299 .expr_as(Expr::col(CratesIoIden::LastModified), Alias::new("date"))
300 .expr_as(
301 Expr::col(CratesIoIden::TotalDownloads),
302 Alias::new("total_downloads"),
303 )
304 .expr_as(
305 Expr::col(CratesIoIden::Description),
306 Alias::new("description"),
307 )
308 .expr_as(
309 Expr::col(CratesIoMetaIden::Documentation),
310 Alias::new("documentation"),
311 )
312 .expr_as(Expr::cust("true"), Alias::new("is_cache"))
313 .from(CratesIoMetaIden::Table)
314 .inner_join(
315 CratesIoIden::Table,
316 Expr::col((CratesIoMetaIden::Table, CratesIoMetaIden::CratesIoFk))
317 .equals((CratesIoIden::Table, CratesIoIden::Id)),
318 )
319 .and_where(
320 Expr::col((CratesIoMetaIden::Table, CratesIoMetaIden::Version))
321 .equals((CratesIoIden::Table, CratesIoIden::MaxVersion)),
322 );
323 if let Some(contains) = contains {
324 query2.and_where(
325 Expr::col((CratesIoIden::Table, CratesIoIden::OriginalName))
326 .like(format!("%{contains}%")),
327 );
328 }
329 query.union(UnionType::All, query2);
330 }
331
332 query.order_by(Alias::new("name"), Order::Asc);
334 if let Some((limit, offset)) = limit_offset {
335 query.limit(limit).offset(offset);
336 }
337
338 let builder = self.db_con.get_database_backend();
339 CrateOverview::find_by_statement(builder.build(&query))
340 .all(&self.db_con)
341 .await
342 .map_err(DbError::from)
343 }
344
345 async fn count<T>(&self, table: T, id_column: T, error: DbError) -> DbResult<u64>
347 where
348 T: Iden + Copy,
349 {
350 #[derive(Debug, PartialEq, FromQueryResult)]
351 struct CountResult {
352 count: Option<i64>,
353 }
354
355 let statement = self.db_con.get_database_backend().build(
356 Query::select()
357 .expr_as(Expr::col((table, id_column)).count(), Alias::new("count"))
358 .from(table),
359 );
360 let Some(result) = CountResult::find_by_statement(statement)
361 .one(&self.db_con)
362 .await?
363 else {
364 return Err(error);
365 };
366 result
367 .count
368 .and_then(|v| u64::try_from(v).ok())
369 .ok_or(error)
370 }
371
372 async fn get_webhook_model(&self, id: &str) -> DbResult<webhook::Model> {
373 webhook::Entity::find()
374 .filter(webhook::Column::Id.eq(
375 TryInto::<Uuid>::try_into(id).map_err(|_| DbError::InvalidId(id.to_string()))?,
376 ))
377 .one(&self.db_con)
378 .await?
379 .ok_or(DbError::WebhookNotFound)
380 }
381
382 async fn get_webhook_queue_model(&self, id: &str) -> DbResult<webhook_queue::Model> {
383 webhook_queue::Entity::find()
384 .filter(webhook_queue::Column::Id.eq(
385 TryInto::<Uuid>::try_into(id).map_err(|_| DbError::InvalidId(id.to_string()))?,
386 ))
387 .one(&self.db_con)
388 .await?
389 .ok_or(DbError::WebhookNotFound)
390 }
391
392 async fn get_owner_by_crate_and_user(
393 &self,
394 crate_name: &str,
395 owner_name: &str,
396 ) -> DbResult<Option<owner::Model>> {
397 owner::Entity::find()
398 .join(JoinType::InnerJoin, owner::Relation::Krate.def())
399 .join(JoinType::InnerJoin, owner::Relation::User.def())
400 .filter(
401 Cond::all()
402 .add(krate::Column::Name.eq(crate_name))
403 .add(user::Column::Name.eq(owner_name)),
404 )
405 .one(&self.db_con)
406 .await
407 .map_err(Into::into)
408 }
409
410 async fn delete_or_not_found<M, A>(&self, model: Option<M>, err: DbError) -> DbResult<()>
411 where
412 M: ModelTrait + sea_orm::IntoActiveModel<A>,
413 A: ActiveModelTrait<Entity = M::Entity> + sea_orm::ActiveModelBehavior + Send,
414 {
415 model.ok_or(err)?.delete(&self.db_con).await?;
416 Ok(())
417 }
418}
419
420fn webhook_model_to_obj(w: webhook::Model) -> DbResult<Webhook> {
421 let event = w
422 .event
423 .as_str()
424 .try_into()
425 .map_err(|_| DbError::InvalidWebhookEvent(w.event))?;
426 Ok(Webhook {
427 id: Some(w.id.into()),
428 name: w.name,
429 event,
430 callback_url: w.callback_url,
431 })
432}
433
434macro_rules! impl_add_relation {
443 ($method:ident, $relation_entity:ident, $insert_entity:ident, $fk1:ident, $fk2:ident) => {
444 async fn $method(&self, fk1: i64, fk2: i64) -> DbResult<()> {
445 let model = $relation_entity::ActiveModel {
446 $fk1: Set(fk1),
447 $fk2: Set(fk2),
448 ..Default::default()
449 };
450 $insert_entity::insert(model).exec(&self.db_con).await?;
451 Ok(())
452 }
453 };
454}
455
456impl Database {
457 impl_add_relation!(
458 add_crate_user_impl,
459 crate_user,
460 CrateUser,
461 crate_fk,
462 user_fk
463 );
464 impl_add_relation!(
465 add_crate_group_impl,
466 crate_group,
467 CrateGroup,
468 crate_fk,
469 group_fk
470 );
471 impl_add_relation!(
472 add_group_user_impl,
473 group_user,
474 GroupUser,
475 group_fk,
476 user_fk
477 );
478 impl_add_relation!(add_owner_impl, owner, Owner, crate_fk, user_fk);
479}
480
481#[async_trait]
482impl DbProvider for Database {
483 async fn get_total_unique_cached_crates(&self) -> DbResult<u64> {
484 self.count(
485 CratesIoIden::Table,
486 CratesIoIden::Id,
487 DbError::FailedToCountCrates,
488 )
489 .await
490 }
491
492 async fn get_total_cached_crate_versions(&self) -> DbResult<u64> {
493 self.count(
494 CratesIoMetaIden::Table,
495 CratesIoMetaIden::Id,
496 DbError::FailedToCountCrateVersions,
497 )
498 .await
499 }
500
501 async fn get_total_cached_downloads(&self) -> DbResult<u64> {
502 #[derive(FromQueryResult)]
503 struct Model {
504 total_downloads: i64,
505 }
506
507 let total_downloads = cratesio_crate::Entity::find()
508 .select_only()
509 .column(cratesio_crate::Column::TotalDownloads)
510 .into_model::<Model>()
511 .all(&self.db_con)
512 .await?;
513
514 Ok(total_downloads
515 .iter()
516 .map(|m| m.total_downloads as u64)
517 .sum())
518 }
519
520 async fn authenticate_user(&self, name: &str, pwd: &str) -> DbResult<User> {
521 let user = self.get_user(name).await?;
522
523 if hash_pwd(pwd, &user.salt) == user.pwd {
524 Ok(user)
525 } else {
526 Err(DbError::PasswordMismatch)
527 }
528 }
529
530 async fn increase_download_counter(
531 &self,
532 crate_name: &NormalizedName,
533 crate_version: &Version,
534 ) -> DbResult<()> {
535 let txn = self.db_con.begin().await?;
536
537 krate::Entity::update_many()
538 .col_expr(
539 krate::Column::TotalDownloads,
540 Expr::col(krate::Column::TotalDownloads).add(1),
541 )
542 .filter(krate::Column::Name.eq(crate_name))
543 .exec(&txn)
544 .await?;
545
546 let crate_id = krate::Entity::find()
547 .filter(krate::Column::Name.eq(crate_name))
548 .one(&txn)
549 .await?
550 .ok_or_else(|| DbError::CrateNotFound(crate_name.to_string()))?
551 .id;
552
553 crate_meta::Entity::update_many()
554 .col_expr(
555 crate_meta::Column::Downloads,
556 Expr::col(crate_meta::Column::Downloads).add(1),
557 )
558 .filter(
559 Cond::all()
560 .add(crate_meta::Column::Version.eq(crate_version))
561 .add(crate_meta::Column::CrateFk.eq(crate_id)),
562 )
563 .exec(&txn)
564 .await?;
565
566 txn.commit().await?;
567 Ok(())
568 }
569
570 async fn increase_cached_download_counter(
571 &self,
572 crate_name: &NormalizedName,
573 crate_version: &Version,
574 ) -> DbResult<()> {
575 let txn = self.db_con.begin().await?;
576
577 cratesio_crate::Entity::update_many()
578 .col_expr(
579 cratesio_crate::Column::TotalDownloads,
580 Expr::col(cratesio_crate::Column::TotalDownloads).add(1),
581 )
582 .filter(cratesio_crate::Column::Name.eq(crate_name))
583 .exec(&txn)
584 .await?;
585
586 let crate_id = cratesio_crate::Entity::find()
587 .filter(cratesio_crate::Column::Name.eq(crate_name))
588 .one(&txn)
589 .await?
590 .ok_or_else(|| DbError::CrateNotFound(crate_name.to_string()))?
591 .id;
592
593 cratesio_meta::Entity::update_many()
594 .col_expr(
595 cratesio_meta::Column::Downloads,
596 Expr::col(cratesio_meta::Column::Downloads).add(1),
597 )
598 .filter(
599 Cond::all()
600 .add(cratesio_meta::Column::Version.eq(crate_version))
601 .add(cratesio_meta::Column::CratesIoFk.eq(crate_id)),
602 )
603 .exec(&txn)
604 .await?;
605
606 txn.commit().await?;
607 Ok(())
608 }
609
610 async fn increase_download_counter_by(
611 &self,
612 crate_name: &NormalizedName,
613 crate_version: &Version,
614 count: u64,
615 ) -> DbResult<()> {
616 let txn = self.db_con.begin().await?;
617
618 krate::Entity::update_many()
619 .col_expr(
620 krate::Column::TotalDownloads,
621 Expr::col(krate::Column::TotalDownloads).add(count as i64),
622 )
623 .filter(krate::Column::Name.eq(crate_name))
624 .exec(&txn)
625 .await?;
626
627 let crate_id = krate::Entity::find()
628 .filter(krate::Column::Name.eq(crate_name))
629 .one(&txn)
630 .await?
631 .ok_or_else(|| DbError::CrateNotFound(crate_name.to_string()))?
632 .id;
633
634 crate_meta::Entity::update_many()
635 .col_expr(
636 crate_meta::Column::Downloads,
637 Expr::col(crate_meta::Column::Downloads).add(count as i64),
638 )
639 .filter(
640 Cond::all()
641 .add(crate_meta::Column::Version.eq(crate_version))
642 .add(crate_meta::Column::CrateFk.eq(crate_id)),
643 )
644 .exec(&txn)
645 .await?;
646
647 txn.commit().await?;
648 Ok(())
649 }
650
651 async fn increase_cached_download_counter_by(
652 &self,
653 crate_name: &NormalizedName,
654 crate_version: &Version,
655 count: u64,
656 ) -> DbResult<()> {
657 let txn = self.db_con.begin().await?;
658
659 cratesio_crate::Entity::update_many()
660 .col_expr(
661 cratesio_crate::Column::TotalDownloads,
662 Expr::col(cratesio_crate::Column::TotalDownloads).add(count as i64),
663 )
664 .filter(cratesio_crate::Column::Name.eq(crate_name))
665 .exec(&txn)
666 .await?;
667
668 let crate_id = cratesio_crate::Entity::find()
669 .filter(cratesio_crate::Column::Name.eq(crate_name))
670 .one(&txn)
671 .await?
672 .ok_or_else(|| DbError::CrateNotFound(crate_name.to_string()))?
673 .id;
674
675 cratesio_meta::Entity::update_many()
676 .col_expr(
677 cratesio_meta::Column::Downloads,
678 Expr::col(cratesio_meta::Column::Downloads).add(count as i64),
679 )
680 .filter(
681 Cond::all()
682 .add(cratesio_meta::Column::Version.eq(crate_version))
683 .add(cratesio_meta::Column::CratesIoFk.eq(crate_id)),
684 )
685 .exec(&txn)
686 .await?;
687
688 txn.commit().await?;
689 Ok(())
690 }
691
692 async fn get_last_updated_crate(&self) -> DbResult<Option<(OriginalName, Version)>> {
693 let krate = krate::Entity::find()
694 .order_by_desc(krate::Column::LastUpdated)
695 .one(&self.db_con)
696 .await?;
697
698 if let Some(krate) = krate {
699 let name = OriginalName::from_unchecked(krate.original_name);
701 let version = Version::from_unchecked_str(&krate.max_version);
703 Ok(Some((name, version)))
704 } else {
705 Ok(None)
706 }
707 }
708
709 async fn validate_session(&self, session_token: &str) -> DbResult<SessionInfo> {
710 let mut query = user::Entity::find()
714 .join(JoinType::InnerJoin, user::Relation::Session.def())
715 .filter(session::Column::Token.eq(session_token));
716 if let Some(cutoff) = self.session_expiry_cutoff() {
717 query = query.filter(session::Column::Created.gte(cutoff));
718 }
719
720 let u = query
721 .one(&self.db_con)
722 .await?
723 .ok_or(DbError::SessionNotFound)?;
724
725 Ok(SessionInfo {
726 name: u.name,
727 is_admin: u.is_admin,
728 is_read_only: u.is_read_only,
729 })
730 }
731
732 async fn delete_expired_sessions(&self) -> DbResult<u64> {
733 let Some(cutoff) = self.session_expiry_cutoff() else {
735 return Ok(0);
736 };
737 let res = session::Entity::delete_many()
738 .filter(session::Column::Created.lt(cutoff))
739 .exec(&self.db_con)
740 .await?;
741 Ok(res.rows_affected)
742 }
743
744 async fn add_session_token(&self, name: &str, session_token: &str) -> DbResult<()> {
745 let user = self.get_user(name).await?;
746 let created = Utc::now().format(DB_DATE_FORMAT).to_string();
747
748 let s = session::ActiveModel {
749 token: Set(session_token.to_owned()),
750 created: Set(created),
751 user_fk: Set(user.id as i64),
752 ..Default::default()
753 };
754
755 s.insert(&self.db_con).await?;
756 Ok(())
757 }
758
759 async fn add_crate_user(&self, crate_name: &NormalizedName, user: &str) -> DbResult<()> {
760 let crate_fk = self.get_krate_model(crate_name).await?.id;
761 let user_fk = self.get_user_model(user).await?.id;
762 self.add_crate_user_impl(crate_fk, user_fk).await
763 }
764
765 async fn add_crate_group(&self, crate_name: &NormalizedName, group: &str) -> DbResult<()> {
766 let crate_fk = self.get_krate_model(crate_name).await?.id;
767 let group_fk = self.get_group_model(group).await?.id;
768 self.add_crate_group_impl(crate_fk, group_fk).await
769 }
770
771 async fn add_group_user(&self, group_name: &str, user: &str) -> DbResult<()> {
772 let group_fk = self.get_group_model(group_name).await?.id;
773 let user_fk = self.get_user_model(user).await?.id;
774 self.add_group_user_impl(group_fk, user_fk).await
775 }
776
777 async fn add_owner(&self, crate_name: &NormalizedName, owner: &str) -> DbResult<()> {
778 let crate_fk = self.get_krate_model(crate_name).await?.id;
779 let user_fk = self.get_user_model(owner).await?.id;
780 self.add_owner_impl(crate_fk, user_fk).await
781 }
782
783 async fn is_download_restricted(&self, crate_name: &NormalizedName) -> DbResult<bool> {
784 Ok(krate::Entity::find()
785 .filter(krate::Column::Name.eq(crate_name))
786 .one(&self.db_con)
787 .await?
788 .is_some_and(|model| model.restricted_download))
789 }
790
791 async fn change_download_restricted(
792 &self,
793 crate_name: &NormalizedName,
794 restricted: bool,
795 ) -> DbResult<()> {
796 let mut krate: krate::ActiveModel = self.get_krate_model(crate_name).await?.into();
797 krate.restricted_download = Set(restricted);
798 krate.update(&self.db_con).await?;
799 Ok(())
800 }
801
802 async fn is_crate_user(&self, crate_name: &NormalizedName, user: &str) -> DbResult<bool> {
803 Ok(self
804 .get_crate_user_by_crate_and_user(crate_name, user)
805 .await?
806 .is_some())
807 }
808
809 async fn is_crate_group(&self, crate_name: &NormalizedName, group: &str) -> DbResult<bool> {
810 Ok(self
811 .get_crate_group_by_name_and_group(crate_name, group)
812 .await?
813 .is_some())
814 }
815
816 async fn is_crate_group_user(&self, crate_name: &NormalizedName, user: &str) -> DbResult<bool> {
817 let user = user::Entity::find()
818 .join(JoinType::InnerJoin, user::Relation::GroupUser.def())
819 .join(JoinType::InnerJoin, group_user::Relation::Group.def())
820 .join(JoinType::InnerJoin, group::Relation::CrateGroup.def())
821 .join(JoinType::InnerJoin, crate_group::Relation::Krate.def())
822 .filter(
823 Cond::all()
824 .add(krate::Column::Name.eq(crate_name))
825 .add(user::Column::Name.eq(user)),
826 )
827 .one(&self.db_con)
828 .await?;
829 Ok(user.is_some())
830 }
831
832 async fn is_group_user(&self, group_name: &str, user: &str) -> DbResult<bool> {
833 Ok(self
834 .get_group_user_by_group_and_user(group_name, user)
835 .await?
836 .is_some())
837 }
838
839 async fn is_owner(&self, crate_name: &NormalizedName, user: &str) -> DbResult<bool> {
840 let owner = owner::Entity::find()
841 .join(JoinType::InnerJoin, owner::Relation::Krate.def())
842 .join(JoinType::InnerJoin, owner::Relation::User.def())
843 .filter(
844 Cond::all()
845 .add(krate::Column::Name.eq(crate_name))
846 .add(user::Column::Name.eq(user)),
847 )
848 .one(&self.db_con)
849 .await?;
850
851 Ok(owner.is_some())
852 }
853
854 async fn get_crate_id(&self, crate_name: &NormalizedName) -> DbResult<Option<i64>> {
855 let id = krate::Entity::find()
856 .filter(krate::Column::Name.eq(crate_name))
857 .one(&self.db_con)
858 .await?
859 .map(|model| model.id);
860
861 Ok(id)
862 }
863
864 async fn get_crate_owners(&self, crate_name: &NormalizedName) -> DbResult<Vec<User>> {
865 let u = user::Entity::find()
866 .join(JoinType::InnerJoin, user::Relation::Owner.def())
867 .join(JoinType::InnerJoin, owner::Relation::Krate.def())
868 .filter(Expr::col((CrateIden::Table, krate::Column::Name)).eq(crate_name))
869 .all(&self.db_con)
870 .await?;
871
872 Ok(u.into_iter().map(User::from).collect())
873 }
874
875 async fn get_crate_users(&self, crate_name: &NormalizedName) -> DbResult<Vec<User>> {
876 let u = user::Entity::find()
877 .join(JoinType::InnerJoin, user::Relation::CrateUser.def())
878 .join(JoinType::InnerJoin, crate_user::Relation::Krate.def())
879 .filter(Expr::col((CrateIden::Table, krate::Column::Name)).eq(crate_name))
880 .all(&self.db_con)
881 .await?;
882
883 Ok(u.into_iter().map(User::from).collect())
884 }
885
886 async fn get_crate_groups(&self, crate_name: &NormalizedName) -> DbResult<Vec<Group>> {
887 let u = group::Entity::find()
888 .join(JoinType::InnerJoin, group::Relation::CrateGroup.def())
889 .join(JoinType::InnerJoin, crate_group::Relation::Krate.def())
890 .filter(Expr::col((CrateIden::Table, krate::Column::Name)).eq(crate_name))
891 .all(&self.db_con)
892 .await?;
893
894 Ok(u.into_iter().map(Group::from).collect())
895 }
896
897 async fn get_group_users(&self, group_name: &str) -> DbResult<Vec<User>> {
898 let u = user::Entity::find()
899 .join(JoinType::InnerJoin, user::Relation::GroupUser.def())
900 .join(JoinType::InnerJoin, group_user::Relation::Group.def())
901 .filter(Expr::col((GroupIden::Table, group::Column::Name)).eq(group_name))
902 .all(&self.db_con)
903 .await?;
904
905 Ok(u.into_iter().map(User::from).collect())
906 }
907
908 async fn get_crate_versions(&self, crate_name: &NormalizedName) -> DbResult<Vec<Version>> {
909 let u = crate_meta::Entity::find()
910 .join(JoinType::InnerJoin, crate_meta::Relation::Krate.def())
911 .filter(Expr::col((CrateIden::Table, krate::Column::Name)).eq(crate_name))
912 .all(&self.db_con)
913 .await?;
914
915 Ok(u.into_iter()
916 .map(|meta| Version::from_unchecked_str(&meta.version))
917 .collect())
918 }
919
920 async fn delete_session_token(&self, session_token: &str) -> DbResult<()> {
921 if let Some(s) = session::Entity::find()
922 .filter(session::Column::Token.eq(session_token))
923 .one(&self.db_con)
924 .await?
925 {
926 s.delete(&self.db_con).await?;
927 }
928
929 Ok(())
930 }
931
932 async fn delete_user(&self, user_name: &str) -> DbResult<()> {
933 self.get_user_model(user_name)
934 .await?
935 .delete(&self.db_con)
936 .await?;
937 Ok(())
938 }
939
940 async fn delete_group(&self, group_name: &str) -> DbResult<()> {
941 self.get_group_model(group_name)
942 .await?
943 .delete(&self.db_con)
944 .await?;
945 Ok(())
946 }
947
948 async fn change_pwd(&self, user_name: &str, new_pwd: &str) -> DbResult<()> {
949 let salt = generate_salt();
950 let hashed = hash_pwd(new_pwd, &salt);
951
952 let mut u: user::ActiveModel = self.get_user_model(user_name).await?.into();
953 u.pwd = Set(hashed);
954 u.salt = Set(salt);
955
956 u.update(&self.db_con).await?;
957 Ok(())
958 }
959
960 async fn change_read_only_state(&self, user_name: &str, state: bool) -> DbResult<()> {
961 let mut u: user::ActiveModel = self.get_user_model(user_name).await?.into();
962 u.is_read_only = Set(state);
963
964 u.update(&self.db_con).await?;
965 Ok(())
966 }
967
968 async fn change_admin_state(&self, user_name: &str, state: bool) -> DbResult<()> {
969 let mut u: user::ActiveModel = self.get_user_model(user_name).await?.into();
970 u.is_admin = Set(state);
971
972 u.update(&self.db_con).await?;
973 Ok(())
974 }
975
976 async fn crate_version_exists(&self, crate_id: i64, version: &str) -> DbResult<bool> {
977 let cm = crate_meta::Entity::find()
978 .filter(
979 Cond::all()
980 .add(crate_meta::Column::CrateFk.eq(crate_id))
981 .add(crate_meta::Column::Version.eq(version)),
982 )
983 .one(&self.db_con)
984 .await?;
985
986 Ok(cm.is_some())
987 }
988
989 async fn get_max_version_from_id(&self, crate_id: i64) -> DbResult<Version> {
990 operations::get_max_version_from_id(&self.db_con, crate_id).await
991 }
992
993 async fn get_max_version_from_name(&self, crate_name: &NormalizedName) -> DbResult<Version> {
994 let k = self.get_krate_model(crate_name).await?;
995 let v = Version::try_from(&k.max_version)
996 .map_err(|_| DbError::FailedToGetMaxVersionByName(crate_name.to_string()))?;
997 Ok(v)
998 }
999
1000 async fn update_max_version(&self, crate_id: i64, version: &Version) -> DbResult<()> {
1001 let krate = krate::Entity::find_by_id(crate_id)
1002 .one(&self.db_con)
1003 .await?
1004 .ok_or(DbError::CrateNotFoundWithId(crate_id))?;
1005
1006 let mut k: krate::ActiveModel = krate.into();
1007 k.max_version = Set(version.to_string());
1008 k.update(&self.db_con).await?;
1009
1010 Ok(())
1011 }
1012
1013 async fn add_auth_token(&self, name: &str, token: &str, user: &str) -> DbResult<()> {
1014 let hashed_token = hash_token(token);
1015 let user = self.get_user_model(user).await?;
1016
1017 let at = auth_token::ActiveModel {
1018 name: Set(name.to_owned()),
1019 token: Set(hashed_token),
1020 user_fk: Set(user.id),
1021 ..Default::default()
1022 };
1023
1024 at.insert(&self.db_con).await?;
1025
1026 Ok(())
1027 }
1028
1029 async fn get_user_from_token(&self, token: &str) -> DbResult<User> {
1030 let token = hash_token(token);
1031
1032 let u = user::Entity::find()
1033 .join(JoinType::InnerJoin, user::Relation::AuthToken.def())
1034 .filter(Expr::col((AuthTokenIden::Table, AuthTokenIden::Token)).eq(token))
1035 .one(&self.db_con)
1036 .await?
1037 .ok_or(DbError::TokenNotFound)?;
1038
1039 Ok(User::from(u))
1040 }
1041
1042 async fn get_user(&self, name: &str) -> DbResult<User> {
1043 Ok(User::from(self.get_user_model(name).await?))
1044 }
1045
1046 async fn get_group(&self, name: &str) -> DbResult<Group> {
1047 Ok(Group::from(self.get_group_model(name).await?))
1048 }
1049
1050 async fn get_auth_tokens(&self, user_name: &str) -> DbResult<Vec<AuthToken>> {
1051 let at: Vec<auth_token::Model> = auth_token::Entity::find()
1052 .join(JoinType::InnerJoin, auth_token::Relation::User.def())
1053 .filter(user::Column::Name.eq(user_name))
1054 .all(&self.db_con)
1055 .await?;
1056
1057 Ok(at.into_iter().map(AuthToken::from).collect())
1058 }
1059
1060 async fn delete_auth_token(&self, id: i32) -> DbResult<()> {
1061 auth_token::Entity::delete_by_id(id as i64)
1062 .exec(&self.db_con)
1063 .await?;
1064 Ok(())
1065 }
1066
1067 async fn delete_owner(&self, crate_name: &str, owner: &str) -> DbResult<()> {
1068 let model = self.get_owner_by_crate_and_user(crate_name, owner).await?;
1069 self.delete_or_not_found(model, DbError::OwnerNotFound(owner.to_string()))
1070 .await
1071 }
1072
1073 async fn delete_crate_user(&self, crate_name: &NormalizedName, user: &str) -> DbResult<()> {
1074 let model = self
1075 .get_crate_user_by_crate_and_user(crate_name, user)
1076 .await?;
1077 self.delete_or_not_found(model, DbError::UserNotFound(user.to_string()))
1078 .await
1079 }
1080
1081 async fn delete_crate_group(&self, crate_name: &NormalizedName, group: &str) -> DbResult<()> {
1082 let model = self
1083 .get_crate_group_by_name_and_group(crate_name, group)
1084 .await?;
1085 self.delete_or_not_found(model, DbError::GroupNotFound(group.to_string()))
1086 .await
1087 }
1088
1089 async fn delete_group_user(&self, group_name: &str, user: &str) -> DbResult<()> {
1090 let model = self
1091 .get_group_user_by_group_and_user(group_name, user)
1092 .await?;
1093 self.delete_or_not_found(model, DbError::UserNotFound(user.to_string()))
1094 .await
1095 }
1096
1097 async fn add_user(
1098 &self,
1099 name: &str,
1100 pwd: &str,
1101 salt: &str,
1102 is_admin: bool,
1103 is_read_only: bool,
1104 ) -> DbResult<()> {
1105 let hashed_pwd = hash_pwd(pwd, salt);
1106 let created = Utc::now().format(DB_DATE_FORMAT).to_string();
1107
1108 let u = user::ActiveModel {
1109 name: Set(name.to_owned()),
1110 pwd: Set(hashed_pwd),
1111 salt: Set(salt.to_owned()),
1112 is_admin: Set(is_admin),
1113 is_read_only: Set(is_read_only),
1114 created: Set(created),
1115 ..Default::default()
1116 };
1117
1118 u.insert(&self.db_con).await?;
1119 Ok(())
1120 }
1121
1122 async fn add_group(&self, name: &str) -> DbResult<()> {
1123 let g = group::ActiveModel {
1124 name: Set(name.to_owned()),
1125 ..Default::default()
1126 };
1127
1128 g.insert(&self.db_con).await?;
1129 Ok(())
1130 }
1131
1132 async fn get_users(&self) -> DbResult<Vec<User>> {
1133 let users = user::Entity::find()
1134 .order_by_asc(user::Column::Name)
1135 .all(&self.db_con)
1136 .await?;
1137
1138 Ok(users.into_iter().map(User::from).collect())
1139 }
1140
1141 async fn get_groups(&self) -> DbResult<Vec<Group>> {
1142 let groups = group::Entity::find()
1143 .order_by_asc(group::Column::Name)
1144 .all(&self.db_con)
1145 .await?;
1146
1147 Ok(groups.into_iter().map(Group::from).collect())
1148 }
1149
1150 async fn get_total_unique_crates(&self) -> DbResult<u64> {
1151 self.count(
1152 CrateIden::Table,
1153 CrateIden::Id,
1154 DbError::FailedToCountCrates,
1155 )
1156 .await
1157 }
1158
1159 async fn get_total_crate_versions(&self) -> DbResult<u64> {
1160 self.count(
1161 CrateMetaIden::Table,
1162 CrateMetaIden::Id,
1163 DbError::FailedToCountCrateVersions,
1164 )
1165 .await
1166 }
1167
1168 async fn get_total_downloads(&self) -> DbResult<u64> {
1169 #[derive(FromQueryResult)]
1170 struct Model {
1171 total_downloads: i64,
1172 }
1173
1174 Ok(krate::Entity::find()
1175 .select_only()
1176 .column(krate::Column::TotalDownloads)
1177 .into_model::<Model>()
1178 .all(&self.db_con)
1179 .await?
1180 .into_iter()
1181 .map(|m| m.total_downloads as u64)
1182 .sum())
1183 }
1184
1185 async fn get_top_crates_downloads(&self, top: u64) -> DbResult<Vec<(String, u64)>> {
1186 #[derive(Debug, PartialEq, FromQueryResult)]
1187 struct SelectResult {
1188 original_name: String,
1189 total_downloads: i64,
1190 }
1191
1192 let stmt = self.db_con.get_database_backend().build(
1193 Query::select()
1194 .columns(vec![CrateIden::OriginalName, CrateIden::TotalDownloads])
1195 .from(CrateIden::Table)
1196 .order_by(CrateIden::TotalDownloads, Order::Desc)
1197 .limit(top),
1198 );
1199
1200 Ok(SelectResult::find_by_statement(stmt)
1201 .all(&self.db_con)
1202 .await?
1203 .into_iter()
1204 .map(|x| (x.original_name, x.total_downloads as u64))
1205 .collect())
1206 }
1207
1208 async fn get_crate_summaries(&self) -> DbResult<Vec<CrateSummary>> {
1209 let krates = krate::Entity::find()
1210 .order_by(krate::Column::Name, Order::Asc)
1211 .all(&self.db_con)
1212 .await?;
1213
1214 Ok(krates.into_iter().map(CrateSummary::from).collect())
1215 }
1216
1217 async fn add_doc_queue(
1218 &self,
1219 krate: &NormalizedName,
1220 version: &Version,
1221 path: &Path,
1222 ) -> DbResult<()> {
1223 let s = doc_queue::ActiveModel {
1224 krate: Set(krate.to_string()),
1225 version: Set(version.to_string()),
1226 path: Set(path.to_string_lossy().to_string()),
1228 ..Default::default()
1229 };
1230
1231 s.insert(&self.db_con).await?;
1232 Ok(())
1233 }
1234
1235 async fn delete_doc_queue(&self, id: i64) -> DbResult<()> {
1236 DocQueue::delete_by_id(id).exec(&self.db_con).await?;
1237 Ok(())
1238 }
1239
1240 async fn get_doc_queue(&self) -> DbResult<Vec<DocQueueEntry>> {
1241 let entities = DocQueue::find().all(&self.db_con).await?;
1242
1243 Ok(entities.into_iter().map(DocQueueEntry::from).collect())
1244 }
1245
1246 async fn delete_crate(&self, krate: &NormalizedName, version: &Version) -> DbResult<()> {
1247 let txn = self.db_con.begin().await?;
1248
1249 let crate_meta_version = crate_meta::Entity::find()
1251 .join(JoinType::InnerJoin, crate_meta::Relation::Krate.def())
1252 .filter(krate::Column::Name.eq(krate))
1253 .filter(crate_meta::Column::Version.eq(version))
1254 .one(&txn)
1255 .await?
1256 .ok_or_else(|| DbError::CrateMetaNotFound(krate.to_string(), version.to_string()))?;
1257 let crate_id = crate_meta_version.crate_fk;
1258 let current_max_version = operations::get_max_version_from_id(&txn, crate_id).await?;
1259 crate_meta_version.delete(&txn).await?;
1260
1261 let crate_index_version = crate_index::Entity::find()
1263 .join(JoinType::InnerJoin, crate_index::Relation::Krate.def())
1264 .filter(krate::Column::Name.eq(krate))
1265 .filter(crate_index::Column::Vers.eq(version))
1266 .one(&txn)
1267 .await?
1268 .ok_or_else(|| DbError::CrateIndexNotFound(krate.to_string(), version.to_string()))?;
1269 crate_index_version.delete(&txn).await?;
1270
1271 let crate_meta_rows = crate_meta::Entity::find()
1274 .join(JoinType::InnerJoin, crate_meta::Relation::Krate.def())
1275 .filter(krate::Column::Name.eq(krate))
1276 .all(&txn)
1277 .await?;
1278
1279 if crate_meta_rows.is_empty() {
1280 krate::Entity::delete_many()
1281 .filter(krate::Column::Name.eq(krate))
1282 .exec(&txn)
1283 .await?;
1284 } else {
1285 let c = krate::Entity::find_by_id(crate_id)
1286 .one(&txn)
1287 .await?
1288 .ok_or(DbError::CrateNotFoundWithId(crate_id))?;
1289 let mut c: krate::ActiveModel = c.into();
1290
1291 if version == ¤t_max_version {
1293 let new_max_version = crate_meta_rows
1294 .into_iter()
1295 .map(|cm| Version::from_unchecked_str(&cm.version))
1296 .max()
1297 .unwrap(); c.max_version = Set(new_max_version.into_inner());
1299 }
1300 let etag = operations::compute_etag(&txn, krate, crate_id).await?;
1302 c.e_tag = Set(etag);
1303 c.update(&txn).await?;
1304 }
1305
1306 txn.commit().await?;
1307
1308 Ok(())
1309 }
1310
1311 async fn get_crate_meta_list(&self, crate_name: &NormalizedName) -> DbResult<Vec<CrateMeta>> {
1312 let crate_meta = crate_meta::Entity::find()
1313 .join(JoinType::InnerJoin, crate_meta::Relation::Krate.def())
1314 .filter(krate::Column::Name.eq(crate_name))
1315 .all(&self.db_con)
1316 .await?;
1317
1318 let crate_meta = crate_meta
1319 .into_iter()
1320 .map(|cm| CrateMeta {
1321 name: crate_name.to_string(),
1322 id: cm.id,
1323 version: cm.version,
1324 created: cm.created,
1325 downloads: cm.downloads,
1326 crate_fk: cm.crate_fk,
1327 })
1328 .collect();
1329
1330 Ok(crate_meta)
1331 }
1332
1333 async fn update_last_updated(&self, id: i64, last_updated: &DateTime<Utc>) -> DbResult<()> {
1334 let krate = krate::Entity::find_by_id(id)
1335 .one(&self.db_con)
1336 .await?
1337 .ok_or(DbError::CrateNotFoundWithId(id))?;
1338
1339 let date = last_updated.format(DB_DATE_FORMAT).to_string();
1340
1341 let mut krate: krate::ActiveModel = krate.into();
1342 krate.last_updated = Set(date);
1343 krate.update(&self.db_con).await?;
1344
1345 Ok(())
1346 }
1347
1348 async fn search_in_crate_name(
1349 &self,
1350 contains: &str,
1351 cache: bool,
1352 ) -> DbResult<Vec<CrateOverview>> {
1353 self.query_crates(Some(contains), None, cache).await
1354 }
1355
1356 async fn get_crate_overview_list(
1357 &self,
1358 limit: u64,
1359 offset: u64,
1360 cache: bool,
1361 ) -> DbResult<Vec<CrateOverview>> {
1362 self.query_crates(None, Some((limit, offset)), cache).await
1363 }
1364
1365 async fn get_crate_data(&self, crate_name: &NormalizedName) -> DbResult<CrateData> {
1366 let krate = self.get_krate_model(crate_name).await?;
1367
1368 let owners: Vec<String> = krate
1369 .find_related(owner::Entity)
1370 .find_also_related(user::Entity)
1371 .all(&self.db_con)
1372 .await?
1373 .into_iter()
1374 .filter_map(|(_, v)| v.map(|v| v.name))
1375 .collect();
1376 let categories: Vec<String> = krate
1377 .find_related(crate_category_to_crate::Entity)
1378 .find_also_related(crate_category::Entity)
1379 .all(&self.db_con)
1380 .await?
1381 .into_iter()
1382 .filter_map(|(_, v)| v.map(|v| v.category))
1383 .collect();
1384 let keywords: Vec<String> = krate
1385 .find_related(crate_keyword_to_crate::Entity)
1386 .find_also_related(crate_keyword::Entity)
1387 .all(&self.db_con)
1388 .await?
1389 .into_iter()
1390 .filter_map(|(_, v)| v.map(|v| v.keyword))
1391 .collect();
1392 let authors: Vec<String> = krate
1393 .find_related(crate_author_to_crate::Entity)
1394 .find_also_related(crate_author::Entity)
1395 .all(&self.db_con)
1396 .await?
1397 .into_iter()
1398 .filter_map(|(_, v)| v.map(|v| v.author))
1399 .collect();
1400 let crate_metas = krate
1401 .find_related(crate_meta::Entity)
1402 .all(&self.db_con)
1403 .await?;
1404 let crate_indices = krate
1405 .find_related(crate_index::Entity)
1406 .all(&self.db_con)
1407 .await?;
1408
1409 let mut versions = Vec::new();
1410 for cm in crate_metas {
1411 let ci = crate_indices
1412 .iter()
1413 .find(|ci| ci.vers == cm.version)
1414 .ok_or_else(|| {
1415 DbError::CrateIndexNotFound(krate.name.clone(), cm.version.clone())
1416 })?;
1417 let dependencies: Vec<CrateRegistryDep> = match ci.deps.clone() {
1418 Some(deps) => {
1419 let ix = serde_json::from_value::<Vec<IndexDep>>(deps)
1420 .map_err(|e| DbError::FailedToConvertToJson(e.to_string()))?;
1421
1422 let mut ft = Vec::new();
1423
1424 let dep_descs = operations::get_desc_for_deps(&self.db_con, ix.iter()).await?;
1425
1426 for dep in &ix {
1427 ft.push(CrateRegistryDep::from_index(
1428 dep_descs.get(&dep.name).cloned(),
1429 dep.clone(),
1430 ));
1431 }
1432
1433 ft
1434 }
1435 None => Vec::default(),
1436 };
1437 let features: BTreeMap<String, Vec<String>> = match ci.features.clone() {
1438 Some(features) => serde_json::from_value::<BTreeMap<String, Vec<String>>>(features)
1439 .map_err(|e| DbError::FailedToConvertToJson(e.to_string()))?,
1440 None => BTreeMap::default(),
1441 };
1442
1443 versions.push(CrateVersionData {
1444 version: cm.version,
1445 created: cm.created,
1446 downloads: cm.downloads,
1447 readme: cm.readme,
1448 license: cm.license,
1449 license_file: cm.license_file,
1450 documentation: cm.documentation,
1451 dependencies,
1452 checksum: ci.cksum.clone(),
1453 features,
1454 yanked: ci.yanked,
1455 links: ci.links.clone(),
1456 v: ci.v,
1457 });
1458 }
1459 versions.sort_by(|a, b| {
1460 Version::from_unchecked_str(&b.version).cmp(&Version::from_unchecked_str(&a.version))
1461 });
1462
1463 let crate_data = CrateData {
1464 name: krate.original_name,
1465 owners,
1466 max_version: krate.max_version,
1467 total_downloads: krate.total_downloads,
1468 last_updated: krate.last_updated,
1469 homepage: krate.homepage,
1470 description: krate.description,
1471 repository: krate.repository,
1472 categories,
1473 keywords,
1474 authors,
1475 versions,
1476 };
1477
1478 Ok(crate_data)
1479 }
1480
1481 async fn add_empty_crate(&self, name: &str, created: &DateTime<Utc>) -> DbResult<i64> {
1482 let created = created.format(DB_DATE_FORMAT).to_string();
1483 let normalized_name = NormalizedName::from(
1484 OriginalName::try_from(name)
1485 .map_err(|_| DbError::InvalidCrateName(name.to_string()))?,
1486 );
1487 let krate = krate::ActiveModel {
1488 id: ActiveValue::default(),
1489 name: Set(normalized_name.to_string()),
1490 original_name: Set(name.to_string()),
1491 max_version: Set("0.0.0".to_string()),
1492 last_updated: Set(created),
1493 total_downloads: Set(0),
1494 homepage: Set(None),
1495 description: Set(None),
1496 repository: Set(None),
1497 e_tag: Set(String::new()), restricted_download: Set(false),
1499 };
1500 Ok(krate.insert(&self.db_con).await?.id)
1501 }
1502
1503 async fn add_crate(
1504 &self,
1505 pub_metadata: &PublishMetadata,
1506 cksum: &str,
1507 created: &DateTime<Utc>,
1508 owner: &str,
1509 ) -> DbResult<i64> {
1510 let created = created.format(DB_DATE_FORMAT).to_string();
1511 let normalized_name = NormalizedName::from(
1512 OriginalName::try_from(&pub_metadata.name)
1513 .map_err(|_| DbError::InvalidCrateName(pub_metadata.name.clone()))?,
1514 );
1515
1516 let existing = krate::Entity::find()
1517 .filter(krate::Column::Name.eq(&pub_metadata.name))
1518 .one(&self.db_con)
1519 .await?;
1520
1521 let txn = self.db_con.begin().await?;
1522
1523 let crate_id = if let Some(krate) = existing {
1524 let krate_id = krate.id;
1525 let max_version = max(
1526 parse_db_version(&krate.max_version)?,
1527 parse_db_version(&pub_metadata.vers)?,
1528 );
1529
1530 let mut krate: krate::ActiveModel = krate.into();
1531 krate.last_updated = Set(created.clone());
1532 krate.max_version = Set(max_version.into_inner());
1533 krate.homepage = Set(pub_metadata.homepage.clone());
1534 krate.description = Set(pub_metadata.description.clone());
1535 krate.repository = Set(pub_metadata.repository.clone());
1536 krate.e_tag = Set(String::new()); krate.update(&txn).await?;
1538 krate_id
1539 } else {
1540 let krate = krate::ActiveModel {
1541 id: ActiveValue::default(),
1542 name: Set(normalized_name.to_string()),
1543 original_name: Set(pub_metadata.name.clone()),
1544 max_version: Set(pub_metadata.vers.clone()),
1545 last_updated: Set(created.clone()),
1546 total_downloads: Set(0),
1547 homepage: Set(pub_metadata.homepage.clone()),
1548 description: Set(pub_metadata.description.clone()),
1549 repository: Set(pub_metadata.repository.clone()),
1550 e_tag: Set(String::new()), restricted_download: Set(false),
1552 };
1553 let krate = krate.insert(&txn).await?;
1554 krate.id
1555 };
1556
1557 operations::add_owner_if_not_exists(&txn, owner, crate_id).await?;
1558 operations::add_crate_metadata(&txn, pub_metadata, &created, crate_id).await?;
1559 operations::add_crate_index(&txn, pub_metadata, cksum, crate_id).await?;
1560 operations::update_etag(&txn, &pub_metadata.name, crate_id).await?;
1561 operations::update_crate_categories(&txn, pub_metadata, crate_id).await?;
1562 operations::update_crate_keywords(&txn, pub_metadata, crate_id).await?;
1563 operations::update_crate_authors(&txn, pub_metadata, crate_id).await?;
1564
1565 txn.commit().await?;
1566 Ok(crate_id)
1567 }
1568
1569 async fn update_docs_link(
1570 &self,
1571 crate_name: &NormalizedName,
1572 version: &Version,
1573 docs_link: &str,
1574 ) -> DbResult<()> {
1575 let (cm, _c) = crate_meta::Entity::find()
1576 .find_also_related(krate::Entity)
1577 .filter(
1578 Cond::all()
1579 .add(krate::Column::Name.eq(crate_name))
1580 .add(crate_meta::Column::Version.eq(version)),
1581 )
1582 .one(&self.db_con)
1583 .await?
1584 .ok_or_else(|| DbError::CrateNotFound(crate_name.to_string()))?;
1585
1586 let mut cm: crate_meta::ActiveModel = cm.into();
1587 cm.documentation = Set(Some(docs_link.to_string()));
1588 cm.update(&self.db_con).await?;
1589 Ok(())
1590 }
1591
1592 async fn add_crate_metadata(
1593 &self,
1594 pub_metadata: &PublishMetadata,
1595 created: &str,
1596 crate_id: i64,
1597 ) -> DbResult<()> {
1598 operations::add_crate_metadata(&self.db_con, pub_metadata, created, crate_id).await
1599 }
1600
1601 async fn get_prefetch_data(&self, crate_name: &str) -> DbResult<Prefetch> {
1602 let mut krate = krate::Entity::find()
1603 .filter(krate::Column::Name.eq(crate_name))
1604 .find_with_related(crate_index::Entity)
1605 .all(&self.db_con)
1606 .await?
1607 .into_iter();
1608
1609 let (Some((krate, crate_indices)), None) = (krate.next(), krate.next()) else {
1611 return Err(DbError::CrateNotFound(crate_name.to_string()));
1612 };
1613
1614 let index_metadata =
1615 operations::crate_index_model_to_index_metadata(crate_name, crate_indices)?;
1616 let data = operations::index_metadata_to_bytes(&index_metadata)?;
1617
1618 Ok(Prefetch {
1619 data,
1620 etag: krate.e_tag,
1621 last_modified: krate.last_updated,
1622 })
1623 }
1624
1625 async fn is_cratesio_cache_up_to_date(
1626 &self,
1627 crate_name: &NormalizedName,
1628 etag: Option<String>,
1629 last_modified: Option<String>,
1630 ) -> DbResult<PrefetchState> {
1631 let Some(krate) = cratesio_crate::Entity::find()
1632 .filter(cratesio_crate::Column::Name.eq(crate_name))
1633 .one(&self.db_con)
1634 .await?
1635 else {
1636 return Ok(PrefetchState::NotFound);
1637 };
1638
1639 let needs_update = match (etag, last_modified) {
1640 (Some(etag), Some(last_modified)) => {
1641 krate.e_tag != etag || krate.last_modified != last_modified
1642 }
1643 (Some(etag), None) => krate.e_tag != etag,
1644 (None, Some(last_modified)) => krate.last_modified != last_modified,
1645 (None, None) => true,
1646 };
1647
1648 if !needs_update {
1649 Ok(PrefetchState::UpToDate)
1650 } else {
1651 let crate_indices = krate
1652 .find_related(cratesio_index::Entity)
1653 .all(&self.db_con)
1654 .await?;
1655 let index_metadata =
1656 operations::cratesio_index_model_to_index_metadata(crate_name, crate_indices)?;
1657 let data = operations::index_metadata_to_bytes(&index_metadata)?;
1658
1659 Ok(PrefetchState::NeedsUpdate(Prefetch {
1660 data,
1661 etag: krate.e_tag.clone(),
1662 last_modified: krate.last_modified,
1663 }))
1664 }
1665 }
1666
1667 async fn add_cratesio_prefetch_data(
1668 &self,
1669 crate_name: &OriginalName,
1670 etag: &str,
1671 last_modified: &str,
1672 description: Option<String>,
1673 indices: &[IndexMetadata],
1674 ) -> DbResult<Prefetch> {
1675 let normalized_name = crate_name.to_normalized();
1676
1677 let max_version = indices
1678 .iter()
1679 .map(|i| Version::from_unchecked_str(&i.vers))
1680 .max()
1681 .ok_or_else(|| DbError::FailedToGetMaxVersionByName(crate_name.to_string()))?;
1682
1683 let txn = self.db_con.begin().await?;
1689
1690 let krate = cratesio_crate::Entity::find()
1691 .filter(cratesio_crate::Column::Name.eq(&normalized_name))
1692 .one(&txn)
1693 .await?;
1694
1695 let krate = if let Some(krate) = krate {
1696 let mut krate: cratesio_crate::ActiveModel = krate.into();
1697 krate.e_tag = Set(etag.to_string());
1698 krate.last_modified = Set(last_modified.to_string());
1699 krate.max_version = Set(max_version.into_inner());
1700 krate.update(&txn).await?
1701 } else {
1702 let krate = cratesio_crate::ActiveModel {
1703 id: ActiveValue::default(),
1704 name: Set(normalized_name.to_string()),
1705 original_name: Set(crate_name.to_string()),
1706 description: Set(description),
1707 e_tag: Set(etag.to_string()),
1708 last_modified: Set(last_modified.to_string()),
1709 total_downloads: Set(0),
1710 max_version: Set(max_version.to_string()),
1711 };
1712 krate.insert(&txn).await?
1713 };
1714
1715 let current_indices = cratesio_index::Entity::find()
1716 .filter(cratesio_index::Column::CratesIoFk.eq(krate.id))
1717 .all(&txn)
1718 .await?;
1719
1720 for index in indices {
1721 if let Some(current_index) = current_indices.iter().find(|ci| index.vers == ci.vers) {
1723 if index.yanked != current_index.yanked {
1724 let mut ci: cratesio_index::ActiveModel = current_index.to_owned().into();
1725 ci.yanked = Set(index.yanked);
1726 ci.update(&txn).await?;
1727 }
1728 } else {
1729 let deps = if index.deps.is_empty() {
1730 None
1731 } else {
1732 let deps = serde_json::to_value(&index.deps)
1733 .map_err(|e| DbError::FailedToConvertToJson(e.to_string()))?;
1734 Some(deps)
1735 };
1736
1737 let features = serde_json::to_value(&index.features)
1738 .map_err(|e| DbError::FailedToConvertToJson(e.to_string()))?;
1739
1740 let features2 = serde_json::to_value(&index.features2)
1741 .map_err(|e| DbError::FailedToConvertToJson(e.to_string()))?;
1742
1743 let new_index = cratesio_index::ActiveModel {
1744 id: ActiveValue::default(),
1745 name: Set(index.name.clone()),
1746 vers: Set(index.vers.clone()),
1747 deps: Set(deps),
1748 cksum: Set(index.cksum.clone()),
1749 features: Set(Some(features)),
1750 features2: Set(Some(features2)),
1751 yanked: Set(index.yanked),
1752 links: Set(index.links.clone()),
1753 pubtime: Set(index.pubtime.map(|dt| dt.naive_utc())),
1754 v: Set(index.v.unwrap_or(1) as i32),
1755 crates_io_fk: Set(krate.id),
1756 };
1757
1758 new_index.insert(&txn).await?;
1759
1760 let meta = cratesio_meta::ActiveModel {
1762 id: ActiveValue::default(),
1763 version: Set(index.vers.clone()),
1764 downloads: Set(0),
1765 crates_io_fk: Set(krate.id),
1766 documentation: Set(Some(format!(
1767 "https://docs.rs/{normalized_name}/{}",
1768 index.vers,
1769 ))),
1770 };
1771
1772 meta.insert(&txn).await?;
1773 }
1774 }
1775
1776 txn.commit().await?;
1777
1778 Ok(Prefetch {
1779 data: operations::index_metadata_to_bytes(indices)?,
1780 etag: etag.to_string(),
1781 last_modified: last_modified.to_string(),
1782 })
1783 }
1784
1785 async fn get_cratesio_index_update_list(&self) -> DbResult<Vec<CratesioPrefetchMsg>> {
1786 let crates = cratesio_crate::Entity::find().all(&self.db_con).await?;
1787 let msgs = crates
1788 .into_iter()
1789 .map(|krate| {
1790 CratesioPrefetchMsg::Update(UpdateData {
1791 name: OriginalName::from_unchecked(krate.original_name),
1792 etag: Some(krate.e_tag),
1793 last_modified: Some(krate.last_modified),
1794 })
1795 })
1796 .collect();
1797 Ok(msgs)
1798 }
1799
1800 async fn unyank_crate(&self, crate_name: &NormalizedName, version: &Version) -> DbResult<()> {
1801 let mut ci: crate_index::ActiveModel = self
1802 .get_crate_index_model(crate_name, version)
1803 .await?
1804 .into();
1805 ci.yanked = Set(false);
1806 ci.save(&self.db_con).await?;
1807 Ok(())
1808 }
1809
1810 async fn yank_crate(&self, crate_name: &NormalizedName, version: &Version) -> DbResult<()> {
1811 let mut ci: crate_index::ActiveModel = self
1812 .get_crate_index_model(crate_name, version)
1813 .await?
1814 .into();
1815 ci.yanked = Set(true);
1816 ci.save(&self.db_con).await?;
1817 Ok(())
1818 }
1819
1820 async fn register_webhook(&self, webhook: Webhook) -> DbResult<String> {
1821 let w = webhook::ActiveModel {
1822 event: Set(Into::<&str>::into(webhook.event).to_string()),
1823 callback_url: Set(webhook.callback_url),
1824 name: Set(webhook.name),
1825 ..Default::default()
1826 };
1827
1828 let w: webhook::Model = w.insert(&self.db_con).await?;
1829 Ok(w.id.to_string())
1830 }
1831
1832 async fn delete_webhook(&self, id: &str) -> DbResult<()> {
1833 self.get_webhook_model(id)
1834 .await?
1835 .delete(&self.db_con)
1836 .await?;
1837 Ok(())
1838 }
1839
1840 async fn get_webhook(&self, id: &str) -> DbResult<Webhook> {
1841 webhook_model_to_obj(self.get_webhook_model(id).await?)
1842 }
1843
1844 async fn get_all_webhooks(&self) -> DbResult<Vec<Webhook>> {
1845 Ok(webhook::Entity::find()
1846 .all(&self.db_con)
1847 .await?
1848 .into_iter()
1849 .filter_map(|w| webhook_model_to_obj(w).ok())
1850 .collect())
1851 }
1852
1853 async fn add_webhook_queue(
1854 &self,
1855 event: WebhookEvent,
1856 payload: serde_json::Value,
1857 ) -> DbResult<()> {
1858 let w = webhook::Entity::find()
1859 .filter(webhook::Column::Event.eq(Into::<&str>::into(event)))
1860 .all(&self.db_con)
1861 .await?;
1862
1863 if w.is_empty() {
1864 return Ok(());
1865 }
1866
1867 let now = Utc::now();
1868
1869 let entries = w.into_iter().map(|w| webhook_queue::ActiveModel {
1870 webhook_fk: Set(w.id),
1871 payload: Set(payload.clone()),
1872 next_attempt: Set(now.into()),
1873 last_attempt: Set(None),
1874 ..Default::default()
1875 });
1876
1877 webhook_queue::Entity::insert_many(entries)
1878 .exec(&self.db_con)
1879 .await?;
1880 Ok(())
1881 }
1882 async fn get_pending_webhook_queue_entries(
1883 &self,
1884 timestamp: DateTime<Utc>,
1885 ) -> DbResult<Vec<WebhookQueue>> {
1886 let w = webhook_queue::Entity::find()
1887 .find_with_related(webhook::Entity)
1888 .filter(webhook_queue::Column::NextAttempt.lte(timestamp))
1889 .all(&self.db_con)
1890 .await?;
1891
1892 Ok(w.into_iter()
1893 .filter_map(|w| {
1894 Some(WebhookQueue {
1895 id: w.0.id.to_string(),
1896 callback_url: w.1.first()?.callback_url.clone(),
1897 payload: w.0.payload,
1898 last_attempt: w.0.last_attempt.map(Into::into),
1899 next_attempt: w.0.next_attempt.into(),
1900 })
1901 })
1902 .collect())
1903 }
1904
1905 async fn update_webhook_queue(
1906 &self,
1907 id: &str,
1908 last_attempt: DateTime<Utc>,
1909 next_attempt: DateTime<Utc>,
1910 ) -> DbResult<()> {
1911 let mut w: webhook_queue::ActiveModel = self.get_webhook_queue_model(id).await?.into();
1912 w.last_attempt = Set(Some(last_attempt.into()));
1913 w.next_attempt = Set(next_attempt.into());
1914 w.update(&self.db_con).await?;
1915 Ok(())
1916 }
1917
1918 async fn delete_webhook_queue(&self, id: &str) -> DbResult<()> {
1919 self.get_webhook_queue_model(id)
1920 .await?
1921 .delete(&self.db_con)
1922 .await?;
1923 Ok(())
1924 }
1925
1926 async fn get_user_by_oauth2_identity(
1929 &self,
1930 issuer: &str,
1931 subject: &str,
1932 ) -> DbResult<Option<User>> {
1933 let identity = oauth2_identity::Entity::find()
1934 .filter(oauth2_identity::Column::ProviderIssuer.eq(issuer))
1935 .filter(oauth2_identity::Column::Subject.eq(subject))
1936 .one(&self.db_con)
1937 .await?;
1938
1939 if let Some(identity) = identity {
1940 let u = user::Entity::find_by_id(identity.user_fk)
1941 .one(&self.db_con)
1942 .await?
1943 .ok_or_else(|| DbError::UserNotFound(format!("user_id={}", identity.user_fk)))?;
1944
1945 Ok(Some(User::from(u)))
1946 } else {
1947 Ok(None)
1948 }
1949 }
1950
1951 async fn create_oauth2_user(
1952 &self,
1953 username: &str,
1954 issuer: &str,
1955 subject: &str,
1956 email: Option<String>,
1957 is_admin: bool,
1958 is_read_only: bool,
1959 ) -> DbResult<User> {
1960 let txn = self.db_con.begin().await?;
1962
1963 let salt = generate_salt();
1966 let random_pwd = Uuid::new_v4().to_string();
1967 let hashed_pwd = hash_pwd(&random_pwd, &salt);
1968 let created = Utc::now().format(DB_DATE_FORMAT).to_string();
1969
1970 let new_user = user::ActiveModel {
1972 name: Set(username.to_string()),
1973 pwd: Set(hashed_pwd),
1974 salt: Set(salt),
1975 is_admin: Set(is_admin),
1976 is_read_only: Set(is_read_only),
1977 created: Set(created.clone()),
1978 ..Default::default()
1979 };
1980
1981 let res = user::Entity::insert(new_user).exec(&txn).await?;
1982 let user_id = res.last_insert_id;
1983
1984 let identity = oauth2_identity::ActiveModel {
1986 user_fk: Set(user_id),
1987 provider_issuer: Set(issuer.to_string()),
1988 subject: Set(subject.to_string()),
1989 email: Set(email),
1990 created: Set(created.clone()),
1991 ..Default::default()
1992 };
1993
1994 oauth2_identity::Entity::insert(identity).exec(&txn).await?;
1995
1996 txn.commit().await?;
1997
1998 Ok(User {
1999 id: user_id as i32,
2000 name: username.to_string(),
2001 pwd: String::new(), salt: String::new(), is_admin,
2004 is_read_only,
2005 created,
2006 })
2007 }
2008
2009 async fn link_oauth2_identity(
2010 &self,
2011 user_id: i64,
2012 issuer: &str,
2013 subject: &str,
2014 email: Option<String>,
2015 ) -> DbResult<()> {
2016 let created = Utc::now().format(DB_DATE_FORMAT).to_string();
2017 let identity = oauth2_identity::ActiveModel {
2018 user_fk: Set(user_id),
2019 provider_issuer: Set(issuer.to_string()),
2020 subject: Set(subject.to_string()),
2021 email: Set(email),
2022 created: Set(created),
2023 ..Default::default()
2024 };
2025
2026 oauth2_identity::Entity::insert(identity)
2027 .exec(&self.db_con)
2028 .await?;
2029
2030 Ok(())
2031 }
2032
2033 async fn store_oauth2_state(
2036 &self,
2037 state: &str,
2038 pkce_verifier: &str,
2039 nonce: &str,
2040 ) -> DbResult<()> {
2041 let created = Utc::now().format(DB_DATE_FORMAT).to_string();
2042 let s = oauth2_state::ActiveModel {
2043 state: Set(state.to_string()),
2044 pkce_verifier: Set(pkce_verifier.to_string()),
2045 nonce: Set(nonce.to_string()),
2046 created: Set(created),
2047 ..Default::default()
2048 };
2049
2050 oauth2_state::Entity::insert(s).exec(&self.db_con).await?;
2051 Ok(())
2052 }
2053
2054 async fn get_and_delete_oauth2_state(&self, state: &str) -> DbResult<Option<OAuth2StateData>> {
2055 let s = oauth2_state::Entity::find()
2056 .filter(oauth2_state::Column::State.eq(state))
2057 .one(&self.db_con)
2058 .await?;
2059
2060 if let Some(s) = s {
2061 let data = OAuth2StateData {
2062 state: s.state.clone(),
2063 pkce_verifier: s.pkce_verifier.clone(),
2064 nonce: s.nonce.clone(),
2065 };
2066
2067 s.delete(&self.db_con).await?;
2069
2070 Ok(Some(data))
2071 } else {
2072 Ok(None)
2073 }
2074 }
2075
2076 async fn cleanup_expired_oauth2_states(&self) -> DbResult<u64> {
2077 let expiry = Utc::now() - chrono::Duration::minutes(10);
2079 let expiry_str = expiry.format(DB_DATE_FORMAT).to_string();
2080
2081 let result = oauth2_state::Entity::delete_many()
2082 .filter(oauth2_state::Column::Created.lt(expiry_str))
2083 .exec(&self.db_con)
2084 .await?;
2085
2086 Ok(result.rows_affected)
2087 }
2088
2089 async fn is_username_available(&self, username: &str) -> DbResult<bool> {
2090 let existing = user::Entity::find()
2091 .filter(user::Column::Name.eq(username))
2092 .one(&self.db_con)
2093 .await?;
2094
2095 Ok(existing.is_none())
2096 }
2097
2098 async fn add_toolchain(
2101 &self,
2102 name: &str,
2103 version: &str,
2104 date: &str,
2105 channel: Option<String>,
2106 ) -> DbResult<i64> {
2107 let created = Utc::now().format(DB_DATE_FORMAT).to_string();
2108
2109 let model = toolchain::ActiveModel {
2110 id: ActiveValue::NotSet,
2111 name: Set(name.to_string()),
2112 version: Set(version.to_string()),
2113 date: Set(date.to_string()),
2114 channel: Set(channel),
2115 created: Set(created),
2116 };
2117
2118 let result = model.insert(&self.db_con).await?;
2119 Ok(result.id)
2120 }
2121
2122 async fn add_toolchain_target(
2123 &self,
2124 toolchain_id: i64,
2125 target: &str,
2126 storage_path: &str,
2127 hash: &str,
2128 size: i64,
2129 ) -> DbResult<i64> {
2130 let model = toolchain_target::ActiveModel {
2131 id: ActiveValue::NotSet,
2132 toolchain_fk: Set(toolchain_id),
2133 target: Set(target.to_string()),
2134 storage_path: Set(storage_path.to_string()),
2135 hash: Set(hash.to_string()),
2136 size: Set(size),
2137 status: Set("processing".to_string()),
2138 };
2139
2140 let result = model.insert(&self.db_con).await?;
2141 Ok(result.id)
2142 }
2143
2144 async fn set_target_status(&self, target_id: i64, status: &str) -> DbResult<()> {
2145 toolchain_target::Entity::update_many()
2146 .col_expr(toolchain_target::Column::Status, Expr::value(status))
2147 .filter(toolchain_target::Column::Id.eq(target_id))
2148 .exec(&self.db_con)
2149 .await?;
2150 Ok(())
2151 }
2152
2153 async fn add_toolchain_component(
2154 &self,
2155 target_id: i64,
2156 name: &str,
2157 storage_path: &str,
2158 hash: &str,
2159 size: i64,
2160 ) -> DbResult<()> {
2161 let model = toolchain_component::ActiveModel {
2162 id: ActiveValue::NotSet,
2163 toolchain_target_fk: Set(target_id),
2164 name: Set(name.to_string()),
2165 storage_path: Set(storage_path.to_string()),
2166 hash: Set(hash.to_string()),
2167 size: Set(size),
2168 };
2169
2170 model.insert(&self.db_con).await?;
2171 Ok(())
2172 }
2173
2174 async fn get_toolchain_by_channel(
2175 &self,
2176 channel: &str,
2177 ) -> DbResult<Option<ToolchainWithTargets>> {
2178 let tc = toolchain::Entity::find()
2179 .filter(toolchain::Column::Channel.eq(channel))
2180 .one(&self.db_con)
2181 .await?;
2182
2183 Ok(match tc {
2184 Some(tc) => Some(self.toolchain_with_targets(tc).await?),
2185 None => None,
2186 })
2187 }
2188
2189 async fn get_toolchain_by_version(
2190 &self,
2191 name: &str,
2192 version: &str,
2193 ) -> DbResult<Option<ToolchainWithTargets>> {
2194 Ok(
2195 match self.get_toolchain_by_name_version(name, version).await? {
2196 Some(tc) => Some(self.toolchain_with_targets(tc).await?),
2197 None => None,
2198 },
2199 )
2200 }
2201
2202 async fn list_toolchains(&self) -> DbResult<Vec<ToolchainWithTargets>> {
2203 let toolchains = toolchain::Entity::find()
2204 .order_by_desc(toolchain::Column::Created)
2205 .all(&self.db_con)
2206 .await?;
2207
2208 let mut result = Vec::with_capacity(toolchains.len());
2209
2210 for tc in toolchains {
2211 result.push(self.toolchain_with_targets(tc).await?);
2212 }
2213
2214 Ok(result)
2215 }
2216
2217 async fn delete_toolchain(&self, name: &str, version: &str) -> DbResult<()> {
2218 toolchain::Entity::delete_many()
2219 .filter(toolchain::Column::Name.eq(name))
2220 .filter(toolchain::Column::Version.eq(version))
2221 .exec(&self.db_con)
2222 .await?;
2223
2224 Ok(())
2225 }
2226
2227 async fn delete_toolchain_target(
2228 &self,
2229 name: &str,
2230 version: &str,
2231 target: &str,
2232 ) -> DbResult<()> {
2233 if let Some(tc) = self.get_toolchain_by_name_version(name, version).await? {
2234 toolchain_target::Entity::delete_many()
2235 .filter(toolchain_target::Column::ToolchainFk.eq(tc.id))
2236 .filter(toolchain_target::Column::Target.eq(target))
2237 .exec(&self.db_con)
2238 .await?;
2239 }
2240
2241 Ok(())
2242 }
2243
2244 async fn set_channel(&self, channel: &str, name: &str, version: &str) -> DbResult<()> {
2245 let toolchains_with_channel = toolchain::Entity::find()
2247 .filter(toolchain::Column::Channel.eq(channel))
2248 .all(&self.db_con)
2249 .await?;
2250
2251 for tc in toolchains_with_channel {
2252 let mut model: toolchain::ActiveModel = tc.into();
2253 model.channel = Set(None);
2254 model.update(&self.db_con).await?;
2255 }
2256
2257 if let Some(tc) = self.get_toolchain_by_name_version(name, version).await? {
2259 let mut model: toolchain::ActiveModel = tc.into();
2260 model.channel = Set(Some(channel.to_string()));
2261 model.update(&self.db_con).await?;
2262 }
2263
2264 Ok(())
2265 }
2266
2267 async fn get_channels(&self) -> DbResult<Vec<ChannelInfo>> {
2268 let toolchains = toolchain::Entity::find()
2269 .filter(toolchain::Column::Channel.is_not_null())
2270 .all(&self.db_con)
2271 .await?;
2272
2273 Ok(toolchains
2274 .into_iter()
2275 .filter_map(|tc| {
2276 tc.channel.map(|channel| ChannelInfo {
2277 name: channel,
2278 version: tc.version,
2279 date: tc.date,
2280 })
2281 })
2282 .collect())
2283 }
2284}
2285
2286fn parse_db_version(value: &str) -> DbResult<Version> {
2287 Version::try_from(value).map_err(|_| DbError::InvalidVersion(value.to_owned()))
2288}
2289
2290#[cfg(test)]
2291mod session_age_tests {
2292 use super::{MAX_SESSION_AGE, session_age_to_duration};
2293
2294 #[test]
2295 fn zero_means_never_expire() {
2296 assert_eq!(session_age_to_duration(std::time::Duration::ZERO), None);
2297 }
2298
2299 #[test]
2300 fn positive_age_is_preserved() {
2301 assert_eq!(
2302 session_age_to_duration(std::time::Duration::from_secs(3600)),
2303 Some(chrono::Duration::seconds(3600))
2304 );
2305 }
2306
2307 #[test]
2308 fn absurd_age_is_clamped_without_panicking() {
2309 let resolved = session_age_to_duration(std::time::Duration::from_secs(u64::MAX))
2311 .expect("non-zero age yields Some");
2312 assert_eq!(
2313 resolved,
2314 chrono::Duration::from_std(MAX_SESSION_AGE).unwrap()
2315 );
2316 }
2317}