Skip to main content

kellnr_db/database/
mod.rs

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
52/// Upper bound on the configured session age. Clamping to this keeps cutoff
53/// arithmetic well within `chrono::Duration`/`DateTime` range, avoiding
54/// overflow panics for absurd configuration values.
55const MAX_SESSION_AGE: std::time::Duration =
56    std::time::Duration::from_secs(100 * 365 * 24 * 60 * 60);
57
58/// Resolve a configured session age into the internal expiry duration.
59///
60/// A value of zero means sessions never expire ([`None`]). Any positive value
61/// is clamped to [`MAX_SESSION_AGE`] so cutoff calculations cannot overflow.
62fn 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    // `clamped` is at most MAX_SESSION_AGE, which fits comfortably in a
68    // `chrono::Duration`, so the conversion cannot fail.
69    Some(chrono::Duration::from_std(clamped).expect("clamped session age fits chrono::Duration"))
70}
71
72pub struct Database {
73    db_con: DatabaseConnection,
74    /// Maximum lifetime of a session before it is treated as expired, or
75    /// `None` if sessions never expire (configured age of zero).
76    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    /// Timestamp (in [`DB_DATE_FORMAT`]) before which a session is expired, or
96    /// `None` when sessions never expire. A session whose `created` value is
97    /// older than the returned cutoff should not validate.
98    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    /// Looks up a user by name; returns the entity model or [`DbError::UserNotFound`].
104    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    /// Looks up a krate by normalized name; returns the entity model or [`DbError::CrateNotFound`].
113    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    /// Looks up a group by name; returns the entity model or [`DbError::GroupNotFound`].
122    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    /// Loads targets (with their components) for a toolchain and returns a `ToolchainWithTargets`.
131    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    /// Looks up a crate index by name and version; returns the entity model or [`DbError::CrateIndexNotFound`].
169    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    /// Looks up a toolchain by name and version; returns `None` if not found.
183    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    /// Looks up a `crate_group` by crate name and group name; returns `None` if not found.
197    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    /// Looks up a `group_user` by group name and user name; returns `None` if not found.
216    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    /// Looks up a `crate_user` by crate name and user name; returns `None` if not found.
235    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    /// Shared implementation for crate overview listing: optional name filter, optional limit/offset, cache union.
254    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        // Optional name filter
287        if let Some(contains) = contains {
288            query.and_where(
289                Expr::col((CrateIden::Table, CrateIden::Name)).like(format!("%{contains}%")),
290            );
291        }
292
293        // UNION with cached crates
294        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        // Ordering and optional limit/offset
333        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    /// Executes a count query `SELECT COUNT(id_column) FROM table` and returns the count as u64.
346    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
434/// Generates an `add_*` method that links two entities via a join table.
435///
436/// # Parameters
437/// - `$method`: The method name to generate (e.g., `add_crate_user_impl`)
438/// - `$relation_entity`: The join table entity module (e.g., `crate_user`)
439/// - `$insert_entity`: The Sea ORM Entity type for insertion (e.g., `CrateUser`)
440/// - `$fk1`: The foreign key field for the first entity
441/// - `$fk2`: The foreign key field for the second entity
442macro_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            // SAFETY: Unchecked is ok, as only valid crate names are inserted into the database
700            let name = OriginalName::from_unchecked(krate.original_name);
701            // SAFETY: Unchecked is ok, as only valid versions are inserted into the database
702            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        // An expired session is treated like a missing one: the age filter
711        // makes it invisible here so callers get the same SessionNotFound.
712        // With expiry disabled (no cutoff) the filter is omitted.
713        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        // Expiry disabled: nothing to remove.
734        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            // FIXME: Convert Path to String properly, handle errors
1227            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        // Delete the entry from the "crate_meta" table
1250        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        // Delete the crate index entry from "crate_index" table
1262        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        // If it was the last entry in the "crate_meta" table, delete the entry
1272        // in the "crate" table as well
1273        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            // Update the max. version if the deleted version was the max. version.
1292            if version == &current_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(); // Safe to unwrap, as crate_meta_rows is not empty
1298                c.max_version = Set(new_max_version.into_inner());
1299            }
1300            // Update the ETag value of the crate index.
1301            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()), // Set to empty string, as it can be computed, when the crate index is inserted
1498            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()); // Set to empty string, as it can be computed, when the crate index is inserted
1537            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()), // Set to empty string, as it can be computed, when the crate index is inserted
1551                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        // Expecting only one crate with its related indices
1610        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        // Use a transaction to ensure the etag update and all index inserts are
1684        // atomic. Without this, a partial failure would leave the crate record
1685        // with the latest etag but incomplete index data, causing stale cache
1686        // entries that can never self-heal (the background updater would get 304
1687        // from crates.io because the etag matches).
1688        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            // Check if the version was yanked or un-yanked and update if so.
1722            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                // Add the meta data for the crate version.
1761                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    // OAuth2 identity methods
1927
1928    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        // Use a transaction to ensure atomicity
1961        let txn = self.db_con.begin().await?;
1962
1963        // Generate a random password and salt for OAuth2 users
1964        // They won't use password auth, but we need valid values
1965        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        // Create the user
1971        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        // Link the OAuth2 identity
1985        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(),  // Don't expose password hash
2002            salt: String::new(), // Don't expose salt
2003            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    // OAuth2 state methods (CSRF/PKCE during auth flow)
2034
2035    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            // Delete after retrieving (single use)
2068            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        // States older than 10 minutes are expired
2078        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    // Toolchain distribution methods
2099
2100    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        // First, clear the channel from any other toolchain that has it
2246        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        // Then set the channel on the target toolchain
2258        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        // u64::MAX seconds would overflow chrono::Duration if not clamped.
2310        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}