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