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