1use async_trait::async_trait;
9use chrono::{DateTime, Utc};
10use mas_data_model::{
11 Clock, UlidExt as _, UpstreamOAuthProvider, UpstreamOAuthProviderClaimsImports,
12};
13use mas_storage::{
14 Page, Pagination,
15 pagination::Node,
16 upstream_oauth2::{
17 UpstreamOAuthProviderFilter, UpstreamOAuthProviderParams, UpstreamOAuthProviderRepository,
18 },
19};
20use opentelemetry_semantic_conventions::attribute::DB_QUERY_TEXT;
21use rand::Rng;
22use sea_query::{Expr, ExprTrait, PostgresQueryBuilder, Query, enum_def};
23use sea_query_sqlx::SqlxBinder;
24use sqlx::{PgConnection, types::Json};
25use tracing::{Instrument, info_span};
26use ulid::Ulid;
27use uuid::Uuid;
28
29use crate::{
30 DatabaseError, DatabaseInconsistencyError,
31 filter::{Filter, StatementExt},
32 iden::UpstreamOAuthProviders,
33 pagination::QueryBuilderExt,
34 tracing::ExecuteExt,
35};
36
37pub struct PgUpstreamOAuthProviderRepository<'c> {
40 conn: &'c mut PgConnection,
41}
42
43impl<'c> PgUpstreamOAuthProviderRepository<'c> {
44 pub fn new(conn: &'c mut PgConnection) -> Self {
47 Self { conn }
48 }
49}
50
51#[derive(sqlx::FromRow)]
52#[enum_def]
53struct ProviderLookup {
54 upstream_oauth_provider_id: Uuid,
55 issuer: Option<String>,
56 human_name: Option<String>,
57 brand_name: Option<String>,
58 scope: String,
59 client_id: String,
60 encrypted_client_secret: Option<String>,
61 token_endpoint_signing_alg: Option<String>,
62 token_endpoint_auth_method: String,
63 id_token_signed_response_alg: String,
64 fetch_userinfo: bool,
65 userinfo_signed_response_alg: Option<String>,
66 created_at: DateTime<Utc>,
67 disabled_at: Option<DateTime<Utc>>,
68 claims_imports: Json<UpstreamOAuthProviderClaimsImports>,
69 jwks_uri_override: Option<String>,
70 authorization_endpoint_override: Option<String>,
71 token_endpoint_override: Option<String>,
72 userinfo_endpoint_override: Option<String>,
73 discovery_mode: String,
74 pkce_mode: String,
75 response_mode: Option<String>,
76 additional_parameters: Option<Json<Vec<(String, String)>>>,
77 forward_login_hint: bool,
78 on_backchannel_logout: String,
79 registration_token_required: bool,
80}
81
82impl Node<Ulid> for ProviderLookup {
83 fn cursor(&self) -> Ulid {
84 self.upstream_oauth_provider_id.into()
85 }
86}
87
88impl TryFrom<ProviderLookup> for UpstreamOAuthProvider {
89 type Error = DatabaseInconsistencyError;
90
91 fn try_from(value: ProviderLookup) -> Result<Self, Self::Error> {
92 let id = value.upstream_oauth_provider_id.into();
93 let scope = value.scope.parse().map_err(|e| {
94 DatabaseInconsistencyError::on("upstream_oauth_providers")
95 .column("scope")
96 .row(id)
97 .source(e)
98 })?;
99 let token_endpoint_auth_method = value.token_endpoint_auth_method.parse().map_err(|e| {
100 DatabaseInconsistencyError::on("upstream_oauth_providers")
101 .column("token_endpoint_auth_method")
102 .row(id)
103 .source(e)
104 })?;
105 let token_endpoint_signing_alg = value
106 .token_endpoint_signing_alg
107 .map(|x| x.parse())
108 .transpose()
109 .map_err(|e| {
110 DatabaseInconsistencyError::on("upstream_oauth_providers")
111 .column("token_endpoint_signing_alg")
112 .row(id)
113 .source(e)
114 })?;
115 let id_token_signed_response_alg =
116 value.id_token_signed_response_alg.parse().map_err(|e| {
117 DatabaseInconsistencyError::on("upstream_oauth_providers")
118 .column("id_token_signed_response_alg")
119 .row(id)
120 .source(e)
121 })?;
122
123 let userinfo_signed_response_alg = value
124 .userinfo_signed_response_alg
125 .map(|x| x.parse())
126 .transpose()
127 .map_err(|e| {
128 DatabaseInconsistencyError::on("upstream_oauth_providers")
129 .column("userinfo_signed_response_alg")
130 .row(id)
131 .source(e)
132 })?;
133
134 let authorization_endpoint_override = value
135 .authorization_endpoint_override
136 .map(|x| x.parse())
137 .transpose()
138 .map_err(|e| {
139 DatabaseInconsistencyError::on("upstream_oauth_providers")
140 .column("authorization_endpoint_override")
141 .row(id)
142 .source(e)
143 })?;
144
145 let token_endpoint_override = value
146 .token_endpoint_override
147 .map(|x| x.parse())
148 .transpose()
149 .map_err(|e| {
150 DatabaseInconsistencyError::on("upstream_oauth_providers")
151 .column("token_endpoint_override")
152 .row(id)
153 .source(e)
154 })?;
155
156 let userinfo_endpoint_override = value
157 .userinfo_endpoint_override
158 .map(|x| x.parse())
159 .transpose()
160 .map_err(|e| {
161 DatabaseInconsistencyError::on("upstream_oauth_providers")
162 .column("userinfo_endpoint_override")
163 .row(id)
164 .source(e)
165 })?;
166
167 let jwks_uri_override = value
168 .jwks_uri_override
169 .map(|x| x.parse())
170 .transpose()
171 .map_err(|e| {
172 DatabaseInconsistencyError::on("upstream_oauth_providers")
173 .column("jwks_uri_override")
174 .row(id)
175 .source(e)
176 })?;
177
178 let discovery_mode = value.discovery_mode.parse().map_err(|e| {
179 DatabaseInconsistencyError::on("upstream_oauth_providers")
180 .column("discovery_mode")
181 .row(id)
182 .source(e)
183 })?;
184
185 let pkce_mode = value.pkce_mode.parse().map_err(|e| {
186 DatabaseInconsistencyError::on("upstream_oauth_providers")
187 .column("pkce_mode")
188 .row(id)
189 .source(e)
190 })?;
191
192 let response_mode = value
193 .response_mode
194 .map(|x| x.parse())
195 .transpose()
196 .map_err(|e| {
197 DatabaseInconsistencyError::on("upstream_oauth_providers")
198 .column("response_mode")
199 .row(id)
200 .source(e)
201 })?;
202
203 let additional_authorization_parameters = value
204 .additional_parameters
205 .map(|Json(x)| x)
206 .unwrap_or_default();
207
208 let on_backchannel_logout = value.on_backchannel_logout.parse().map_err(|e| {
209 DatabaseInconsistencyError::on("upstream_oauth_providers")
210 .column("on_backchannel_logout")
211 .row(id)
212 .source(e)
213 })?;
214
215 Ok(UpstreamOAuthProvider {
216 id,
217 issuer: value.issuer,
218 human_name: value.human_name,
219 brand_name: value.brand_name,
220 scope,
221 client_id: value.client_id,
222 encrypted_client_secret: value.encrypted_client_secret,
223 token_endpoint_auth_method,
224 token_endpoint_signing_alg,
225 id_token_signed_response_alg,
226 fetch_userinfo: value.fetch_userinfo,
227 userinfo_signed_response_alg,
228 created_at: value.created_at,
229 disabled_at: value.disabled_at,
230 claims_imports: value.claims_imports.0,
231 authorization_endpoint_override,
232 token_endpoint_override,
233 userinfo_endpoint_override,
234 jwks_uri_override,
235 discovery_mode,
236 pkce_mode,
237 response_mode,
238 additional_authorization_parameters,
239 forward_login_hint: value.forward_login_hint,
240 on_backchannel_logout,
241 registration_token_required: value.registration_token_required,
242 })
243 }
244}
245
246impl Filter for UpstreamOAuthProviderFilter<'_> {
247 fn generate_condition(&self, _has_joins: bool) -> impl sea_query::IntoCondition {
248 sea_query::Condition::all().add_option(self.enabled().map(|enabled| {
249 Expr::col((
250 UpstreamOAuthProviders::Table,
251 UpstreamOAuthProviders::DisabledAt,
252 ))
253 .is_null()
254 .eq(enabled)
255 }))
256 }
257}
258
259#[async_trait]
260impl UpstreamOAuthProviderRepository for PgUpstreamOAuthProviderRepository<'_> {
261 type Error = DatabaseError;
262
263 #[tracing::instrument(
264 name = "db.upstream_oauth_provider.lookup",
265 skip_all,
266 fields(
267 db.query.text,
268 upstream_oauth_provider.id = %id,
269 ),
270 )]
271 async fn lookup(&mut self, id: Ulid) -> Result<Option<UpstreamOAuthProvider>, Self::Error> {
272 let res = sqlx::query_as!(
273 ProviderLookup,
274 r#"
275 SELECT
276 upstream_oauth_provider_id,
277 issuer,
278 human_name,
279 brand_name,
280 scope,
281 client_id,
282 encrypted_client_secret,
283 token_endpoint_signing_alg,
284 token_endpoint_auth_method,
285 id_token_signed_response_alg,
286 fetch_userinfo,
287 userinfo_signed_response_alg,
288 created_at,
289 disabled_at,
290 claims_imports as "claims_imports: Json<UpstreamOAuthProviderClaimsImports>",
291 jwks_uri_override,
292 authorization_endpoint_override,
293 token_endpoint_override,
294 userinfo_endpoint_override,
295 discovery_mode,
296 pkce_mode,
297 response_mode,
298 additional_parameters as "additional_parameters: Json<Vec<(String, String)>>",
299 forward_login_hint,
300 on_backchannel_logout,
301 registration_token_required
302 FROM upstream_oauth_providers
303 WHERE upstream_oauth_provider_id = $1
304 "#,
305 Uuid::from(id),
306 )
307 .traced()
308 .fetch_optional(&mut *self.conn)
309 .await?;
310
311 let res = res
312 .map(UpstreamOAuthProvider::try_from)
313 .transpose()
314 .map_err(DatabaseError::from)?;
315
316 Ok(res)
317 }
318
319 #[tracing::instrument(
320 name = "db.upstream_oauth_provider.add",
321 skip_all,
322 fields(
323 db.query.text,
324 upstream_oauth_provider.id,
325 upstream_oauth_provider.issuer = params.issuer,
326 upstream_oauth_provider.client_id = %params.client_id,
327 ),
328 )]
329 async fn add(
330 &mut self,
331 rng: &mut (dyn Rng + Send),
332 clock: &dyn Clock,
333 params: UpstreamOAuthProviderParams,
334 ) -> Result<UpstreamOAuthProvider, Self::Error> {
335 let created_at = clock.now();
336 let id = Ulid::from_datetime_with_rng(created_at, rng);
337 tracing::Span::current().record("upstream_oauth_provider.id", tracing::field::display(id));
338
339 sqlx::query!(
340 r#"
341 INSERT INTO upstream_oauth_providers (
342 upstream_oauth_provider_id,
343 issuer,
344 human_name,
345 brand_name,
346 scope,
347 token_endpoint_auth_method,
348 token_endpoint_signing_alg,
349 id_token_signed_response_alg,
350 fetch_userinfo,
351 userinfo_signed_response_alg,
352 client_id,
353 encrypted_client_secret,
354 claims_imports,
355 authorization_endpoint_override,
356 token_endpoint_override,
357 userinfo_endpoint_override,
358 jwks_uri_override,
359 discovery_mode,
360 pkce_mode,
361 response_mode,
362 forward_login_hint,
363 on_backchannel_logout,
364 registration_token_required,
365 created_at
366 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11,
367 $12, $13, $14, $15, $16, $17, $18, $19, $20,
368 $21, $22, $23, $24)
369 "#,
370 Uuid::from(id),
371 params.issuer.as_deref(),
372 params.human_name.as_deref(),
373 params.brand_name.as_deref(),
374 params.scope.to_string(),
375 params.token_endpoint_auth_method.to_string(),
376 params
377 .token_endpoint_signing_alg
378 .as_ref()
379 .map(ToString::to_string),
380 params.id_token_signed_response_alg.to_string(),
381 params.fetch_userinfo,
382 params
383 .userinfo_signed_response_alg
384 .as_ref()
385 .map(ToString::to_string),
386 ¶ms.client_id,
387 params.encrypted_client_secret.as_deref(),
388 Json(¶ms.claims_imports) as _,
389 params
390 .authorization_endpoint_override
391 .as_ref()
392 .map(ToString::to_string),
393 params
394 .token_endpoint_override
395 .as_ref()
396 .map(ToString::to_string),
397 params
398 .userinfo_endpoint_override
399 .as_ref()
400 .map(ToString::to_string),
401 params.jwks_uri_override.as_ref().map(ToString::to_string),
402 params.discovery_mode.as_str(),
403 params.pkce_mode.as_str(),
404 params.response_mode.as_ref().map(ToString::to_string),
405 params.forward_login_hint,
406 params.on_backchannel_logout.as_str(),
407 params.registration_token_required,
408 created_at,
409 )
410 .traced()
411 .execute(&mut *self.conn)
412 .await?;
413
414 Ok(UpstreamOAuthProvider {
415 id,
416 issuer: params.issuer,
417 human_name: params.human_name,
418 brand_name: params.brand_name,
419 scope: params.scope,
420 client_id: params.client_id,
421 encrypted_client_secret: params.encrypted_client_secret,
422 token_endpoint_signing_alg: params.token_endpoint_signing_alg,
423 token_endpoint_auth_method: params.token_endpoint_auth_method,
424 id_token_signed_response_alg: params.id_token_signed_response_alg,
425 fetch_userinfo: params.fetch_userinfo,
426 userinfo_signed_response_alg: params.userinfo_signed_response_alg,
427 created_at,
428 disabled_at: None,
429 claims_imports: params.claims_imports,
430 authorization_endpoint_override: params.authorization_endpoint_override,
431 token_endpoint_override: params.token_endpoint_override,
432 userinfo_endpoint_override: params.userinfo_endpoint_override,
433 jwks_uri_override: params.jwks_uri_override,
434 discovery_mode: params.discovery_mode,
435 pkce_mode: params.pkce_mode,
436 response_mode: params.response_mode,
437 additional_authorization_parameters: params.additional_authorization_parameters,
438 on_backchannel_logout: params.on_backchannel_logout,
439 forward_login_hint: params.forward_login_hint,
440 registration_token_required: params.registration_token_required,
441 })
442 }
443
444 #[tracing::instrument(
445 name = "db.upstream_oauth_provider.delete_by_id",
446 skip_all,
447 fields(
448 db.query.text,
449 upstream_oauth_provider.id = %id,
450 ),
451 )]
452 async fn delete_by_id(&mut self, id: Ulid) -> Result<(), Self::Error> {
453 {
456 let span = info_span!(
457 "db.oauth2_client.delete_by_id.authorization_sessions",
458 upstream_oauth_provider.id = %id,
459 { DB_QUERY_TEXT } = tracing::field::Empty,
460 );
461 sqlx::query!(
462 r#"
463 DELETE FROM upstream_oauth_authorization_sessions
464 WHERE upstream_oauth_provider_id = $1
465 "#,
466 Uuid::from(id),
467 )
468 .record(&span)
469 .execute(&mut *self.conn)
470 .instrument(span)
471 .await?;
472 }
473
474 {
477 let span = info_span!(
478 "db.oauth2_client.delete_by_id.links",
479 upstream_oauth_provider.id = %id,
480 { DB_QUERY_TEXT } = tracing::field::Empty,
481 );
482 sqlx::query!(
483 r#"
484 DELETE FROM upstream_oauth_links
485 WHERE upstream_oauth_provider_id = $1
486 "#,
487 Uuid::from(id),
488 )
489 .record(&span)
490 .execute(&mut *self.conn)
491 .instrument(span)
492 .await?;
493 }
494
495 let res = sqlx::query!(
496 r#"
497 DELETE FROM upstream_oauth_providers
498 WHERE upstream_oauth_provider_id = $1
499 "#,
500 Uuid::from(id),
501 )
502 .traced()
503 .execute(&mut *self.conn)
504 .await?;
505
506 DatabaseError::ensure_affected_rows(&res, 1)
507 }
508
509 #[tracing::instrument(
510 name = "db.upstream_oauth_provider.add",
511 skip_all,
512 fields(
513 db.query.text,
514 upstream_oauth_provider.id = %id,
515 upstream_oauth_provider.issuer = params.issuer,
516 upstream_oauth_provider.client_id = %params.client_id,
517 ),
518 )]
519 async fn upsert(
520 &mut self,
521 clock: &dyn Clock,
522 id: Ulid,
523 params: UpstreamOAuthProviderParams,
524 ) -> Result<UpstreamOAuthProvider, Self::Error> {
525 let created_at = clock.now();
526
527 let created_at = sqlx::query_scalar!(
528 r#"
529 INSERT INTO upstream_oauth_providers (
530 upstream_oauth_provider_id,
531 issuer,
532 human_name,
533 brand_name,
534 scope,
535 token_endpoint_auth_method,
536 token_endpoint_signing_alg,
537 id_token_signed_response_alg,
538 fetch_userinfo,
539 userinfo_signed_response_alg,
540 client_id,
541 encrypted_client_secret,
542 claims_imports,
543 authorization_endpoint_override,
544 token_endpoint_override,
545 userinfo_endpoint_override,
546 jwks_uri_override,
547 discovery_mode,
548 pkce_mode,
549 response_mode,
550 additional_parameters,
551 forward_login_hint,
552 ui_order,
553 on_backchannel_logout,
554 registration_token_required,
555 created_at
556 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
557 $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
558 $21, $22, $23, $24, $25, $26)
559 ON CONFLICT (upstream_oauth_provider_id)
560 DO UPDATE
561 SET
562 issuer = EXCLUDED.issuer,
563 human_name = EXCLUDED.human_name,
564 brand_name = EXCLUDED.brand_name,
565 scope = EXCLUDED.scope,
566 token_endpoint_auth_method = EXCLUDED.token_endpoint_auth_method,
567 token_endpoint_signing_alg = EXCLUDED.token_endpoint_signing_alg,
568 id_token_signed_response_alg = EXCLUDED.id_token_signed_response_alg,
569 fetch_userinfo = EXCLUDED.fetch_userinfo,
570 userinfo_signed_response_alg = EXCLUDED.userinfo_signed_response_alg,
571 disabled_at = NULL,
572 client_id = EXCLUDED.client_id,
573 encrypted_client_secret = EXCLUDED.encrypted_client_secret,
574 claims_imports = EXCLUDED.claims_imports,
575 authorization_endpoint_override = EXCLUDED.authorization_endpoint_override,
576 token_endpoint_override = EXCLUDED.token_endpoint_override,
577 userinfo_endpoint_override = EXCLUDED.userinfo_endpoint_override,
578 jwks_uri_override = EXCLUDED.jwks_uri_override,
579 discovery_mode = EXCLUDED.discovery_mode,
580 pkce_mode = EXCLUDED.pkce_mode,
581 response_mode = EXCLUDED.response_mode,
582 additional_parameters = EXCLUDED.additional_parameters,
583 forward_login_hint = EXCLUDED.forward_login_hint,
584 ui_order = EXCLUDED.ui_order,
585 on_backchannel_logout = EXCLUDED.on_backchannel_logout,
586 registration_token_required = EXCLUDED.registration_token_required
587 RETURNING created_at
588 "#,
589 Uuid::from(id),
590 params.issuer.as_deref(),
591 params.human_name.as_deref(),
592 params.brand_name.as_deref(),
593 params.scope.to_string(),
594 params.token_endpoint_auth_method.to_string(),
595 params
596 .token_endpoint_signing_alg
597 .as_ref()
598 .map(ToString::to_string),
599 params.id_token_signed_response_alg.to_string(),
600 params.fetch_userinfo,
601 params
602 .userinfo_signed_response_alg
603 .as_ref()
604 .map(ToString::to_string),
605 ¶ms.client_id,
606 params.encrypted_client_secret.as_deref(),
607 Json(¶ms.claims_imports) as _,
608 params
609 .authorization_endpoint_override
610 .as_ref()
611 .map(ToString::to_string),
612 params
613 .token_endpoint_override
614 .as_ref()
615 .map(ToString::to_string),
616 params
617 .userinfo_endpoint_override
618 .as_ref()
619 .map(ToString::to_string),
620 params.jwks_uri_override.as_ref().map(ToString::to_string),
621 params.discovery_mode.as_str(),
622 params.pkce_mode.as_str(),
623 params.response_mode.as_ref().map(ToString::to_string),
624 Json(¶ms.additional_authorization_parameters) as _,
625 params.forward_login_hint,
626 params.ui_order,
627 params.on_backchannel_logout.as_str(),
628 params.registration_token_required,
629 created_at,
630 )
631 .traced()
632 .fetch_one(&mut *self.conn)
633 .await?;
634
635 Ok(UpstreamOAuthProvider {
636 id,
637 issuer: params.issuer,
638 human_name: params.human_name,
639 brand_name: params.brand_name,
640 scope: params.scope,
641 client_id: params.client_id,
642 encrypted_client_secret: params.encrypted_client_secret,
643 token_endpoint_signing_alg: params.token_endpoint_signing_alg,
644 token_endpoint_auth_method: params.token_endpoint_auth_method,
645 id_token_signed_response_alg: params.id_token_signed_response_alg,
646 fetch_userinfo: params.fetch_userinfo,
647 userinfo_signed_response_alg: params.userinfo_signed_response_alg,
648 created_at,
649 disabled_at: None,
650 claims_imports: params.claims_imports,
651 authorization_endpoint_override: params.authorization_endpoint_override,
652 token_endpoint_override: params.token_endpoint_override,
653 userinfo_endpoint_override: params.userinfo_endpoint_override,
654 jwks_uri_override: params.jwks_uri_override,
655 discovery_mode: params.discovery_mode,
656 pkce_mode: params.pkce_mode,
657 response_mode: params.response_mode,
658 additional_authorization_parameters: params.additional_authorization_parameters,
659 forward_login_hint: params.forward_login_hint,
660 on_backchannel_logout: params.on_backchannel_logout,
661 registration_token_required: params.registration_token_required,
662 })
663 }
664
665 #[tracing::instrument(
666 name = "db.upstream_oauth_provider.disable",
667 skip_all,
668 fields(
669 db.query.text,
670 %upstream_oauth_provider.id,
671 ),
672 )]
673 async fn disable(
674 &mut self,
675 clock: &dyn Clock,
676 mut upstream_oauth_provider: UpstreamOAuthProvider,
677 ) -> Result<UpstreamOAuthProvider, Self::Error> {
678 let disabled_at = clock.now();
679 let res = sqlx::query!(
680 r#"
681 UPDATE upstream_oauth_providers
682 SET disabled_at = $2
683 WHERE upstream_oauth_provider_id = $1
684 "#,
685 Uuid::from(upstream_oauth_provider.id),
686 disabled_at,
687 )
688 .traced()
689 .execute(&mut *self.conn)
690 .await?;
691
692 DatabaseError::ensure_affected_rows(&res, 1)?;
693
694 upstream_oauth_provider.disabled_at = Some(disabled_at);
695
696 Ok(upstream_oauth_provider)
697 }
698
699 #[tracing::instrument(
700 name = "db.upstream_oauth_provider.list",
701 skip_all,
702 fields(
703 db.query.text,
704 ),
705 )]
706 async fn list(
707 &mut self,
708 filter: UpstreamOAuthProviderFilter<'_>,
709 pagination: Pagination,
710 ) -> Result<Page<UpstreamOAuthProvider>, Self::Error> {
711 let (sql, arguments) = Query::select()
712 .expr_as(
713 Expr::col((
714 UpstreamOAuthProviders::Table,
715 UpstreamOAuthProviders::UpstreamOAuthProviderId,
716 )),
717 ProviderLookupIden::UpstreamOauthProviderId,
718 )
719 .expr_as(
720 Expr::col((
721 UpstreamOAuthProviders::Table,
722 UpstreamOAuthProviders::Issuer,
723 )),
724 ProviderLookupIden::Issuer,
725 )
726 .expr_as(
727 Expr::col((
728 UpstreamOAuthProviders::Table,
729 UpstreamOAuthProviders::HumanName,
730 )),
731 ProviderLookupIden::HumanName,
732 )
733 .expr_as(
734 Expr::col((
735 UpstreamOAuthProviders::Table,
736 UpstreamOAuthProviders::BrandName,
737 )),
738 ProviderLookupIden::BrandName,
739 )
740 .expr_as(
741 Expr::col((UpstreamOAuthProviders::Table, UpstreamOAuthProviders::Scope)),
742 ProviderLookupIden::Scope,
743 )
744 .expr_as(
745 Expr::col((
746 UpstreamOAuthProviders::Table,
747 UpstreamOAuthProviders::ClientId,
748 )),
749 ProviderLookupIden::ClientId,
750 )
751 .expr_as(
752 Expr::col((
753 UpstreamOAuthProviders::Table,
754 UpstreamOAuthProviders::EncryptedClientSecret,
755 )),
756 ProviderLookupIden::EncryptedClientSecret,
757 )
758 .expr_as(
759 Expr::col((
760 UpstreamOAuthProviders::Table,
761 UpstreamOAuthProviders::TokenEndpointSigningAlg,
762 )),
763 ProviderLookupIden::TokenEndpointSigningAlg,
764 )
765 .expr_as(
766 Expr::col((
767 UpstreamOAuthProviders::Table,
768 UpstreamOAuthProviders::TokenEndpointAuthMethod,
769 )),
770 ProviderLookupIden::TokenEndpointAuthMethod,
771 )
772 .expr_as(
773 Expr::col((
774 UpstreamOAuthProviders::Table,
775 UpstreamOAuthProviders::IdTokenSignedResponseAlg,
776 )),
777 ProviderLookupIden::IdTokenSignedResponseAlg,
778 )
779 .expr_as(
780 Expr::col((
781 UpstreamOAuthProviders::Table,
782 UpstreamOAuthProviders::FetchUserinfo,
783 )),
784 ProviderLookupIden::FetchUserinfo,
785 )
786 .expr_as(
787 Expr::col((
788 UpstreamOAuthProviders::Table,
789 UpstreamOAuthProviders::UserinfoSignedResponseAlg,
790 )),
791 ProviderLookupIden::UserinfoSignedResponseAlg,
792 )
793 .expr_as(
794 Expr::col((
795 UpstreamOAuthProviders::Table,
796 UpstreamOAuthProviders::CreatedAt,
797 )),
798 ProviderLookupIden::CreatedAt,
799 )
800 .expr_as(
801 Expr::col((
802 UpstreamOAuthProviders::Table,
803 UpstreamOAuthProviders::DisabledAt,
804 )),
805 ProviderLookupIden::DisabledAt,
806 )
807 .expr_as(
808 Expr::col((
809 UpstreamOAuthProviders::Table,
810 UpstreamOAuthProviders::ClaimsImports,
811 )),
812 ProviderLookupIden::ClaimsImports,
813 )
814 .expr_as(
815 Expr::col((
816 UpstreamOAuthProviders::Table,
817 UpstreamOAuthProviders::JwksUriOverride,
818 )),
819 ProviderLookupIden::JwksUriOverride,
820 )
821 .expr_as(
822 Expr::col((
823 UpstreamOAuthProviders::Table,
824 UpstreamOAuthProviders::TokenEndpointOverride,
825 )),
826 ProviderLookupIden::TokenEndpointOverride,
827 )
828 .expr_as(
829 Expr::col((
830 UpstreamOAuthProviders::Table,
831 UpstreamOAuthProviders::AuthorizationEndpointOverride,
832 )),
833 ProviderLookupIden::AuthorizationEndpointOverride,
834 )
835 .expr_as(
836 Expr::col((
837 UpstreamOAuthProviders::Table,
838 UpstreamOAuthProviders::UserinfoEndpointOverride,
839 )),
840 ProviderLookupIden::UserinfoEndpointOverride,
841 )
842 .expr_as(
843 Expr::col((
844 UpstreamOAuthProviders::Table,
845 UpstreamOAuthProviders::DiscoveryMode,
846 )),
847 ProviderLookupIden::DiscoveryMode,
848 )
849 .expr_as(
850 Expr::col((
851 UpstreamOAuthProviders::Table,
852 UpstreamOAuthProviders::PkceMode,
853 )),
854 ProviderLookupIden::PkceMode,
855 )
856 .expr_as(
857 Expr::col((
858 UpstreamOAuthProviders::Table,
859 UpstreamOAuthProviders::ResponseMode,
860 )),
861 ProviderLookupIden::ResponseMode,
862 )
863 .expr_as(
864 Expr::col((
865 UpstreamOAuthProviders::Table,
866 UpstreamOAuthProviders::AdditionalParameters,
867 )),
868 ProviderLookupIden::AdditionalParameters,
869 )
870 .expr_as(
871 Expr::col((
872 UpstreamOAuthProviders::Table,
873 UpstreamOAuthProviders::ForwardLoginHint,
874 )),
875 ProviderLookupIden::ForwardLoginHint,
876 )
877 .expr_as(
878 Expr::col((
879 UpstreamOAuthProviders::Table,
880 UpstreamOAuthProviders::OnBackchannelLogout,
881 )),
882 ProviderLookupIden::OnBackchannelLogout,
883 )
884 .expr_as(
885 Expr::col((
886 UpstreamOAuthProviders::Table,
887 UpstreamOAuthProviders::RegistrationTokenRequired,
888 )),
889 ProviderLookupIden::RegistrationTokenRequired,
890 )
891 .from(UpstreamOAuthProviders::Table)
892 .apply_filter(filter)
893 .generate_pagination(
894 (
895 UpstreamOAuthProviders::Table,
896 UpstreamOAuthProviders::UpstreamOAuthProviderId,
897 ),
898 pagination,
899 )
900 .build_sqlx(PostgresQueryBuilder);
901
902 let edges: Vec<ProviderLookup> = sqlx::query_as_with(&sql, arguments)
903 .traced()
904 .fetch_all(&mut *self.conn)
905 .await?;
906
907 let page = pagination
908 .process(edges)
909 .try_map(UpstreamOAuthProvider::try_from)?;
910
911 return Ok(page);
912 }
913
914 #[tracing::instrument(
915 name = "db.upstream_oauth_provider.count",
916 skip_all,
917 fields(
918 db.query.text,
919 ),
920 )]
921 async fn count(
922 &mut self,
923 filter: UpstreamOAuthProviderFilter<'_>,
924 ) -> Result<usize, Self::Error> {
925 let (sql, arguments) = Query::select()
926 .expr(
927 Expr::col((
928 UpstreamOAuthProviders::Table,
929 UpstreamOAuthProviders::UpstreamOAuthProviderId,
930 ))
931 .count(),
932 )
933 .from(UpstreamOAuthProviders::Table)
934 .apply_filter(filter)
935 .build_sqlx(PostgresQueryBuilder);
936
937 let count: i64 = sqlx::query_scalar_with(&sql, arguments)
938 .traced()
939 .fetch_one(&mut *self.conn)
940 .await?;
941
942 count
943 .try_into()
944 .map_err(DatabaseError::to_invalid_operation)
945 }
946
947 #[tracing::instrument(
948 name = "db.upstream_oauth_provider.all_enabled",
949 skip_all,
950 fields(
951 db.query.text,
952 ),
953 )]
954 async fn all_enabled(&mut self) -> Result<Vec<UpstreamOAuthProvider>, Self::Error> {
955 let res = sqlx::query_as!(
956 ProviderLookup,
957 r#"
958 SELECT
959 upstream_oauth_provider_id,
960 issuer,
961 human_name,
962 brand_name,
963 scope,
964 client_id,
965 encrypted_client_secret,
966 token_endpoint_signing_alg,
967 token_endpoint_auth_method,
968 id_token_signed_response_alg,
969 fetch_userinfo,
970 userinfo_signed_response_alg,
971 created_at,
972 disabled_at,
973 claims_imports as "claims_imports: Json<UpstreamOAuthProviderClaimsImports>",
974 jwks_uri_override,
975 authorization_endpoint_override,
976 token_endpoint_override,
977 userinfo_endpoint_override,
978 discovery_mode,
979 pkce_mode,
980 response_mode,
981 additional_parameters as "additional_parameters: Json<Vec<(String, String)>>",
982 forward_login_hint,
983 on_backchannel_logout,
984 registration_token_required
985
986 FROM upstream_oauth_providers
987 WHERE disabled_at IS NULL
988 ORDER BY ui_order ASC, upstream_oauth_provider_id ASC
989 "#,
990 )
991 .traced()
992 .fetch_all(&mut *self.conn)
993 .await?;
994
995 let res: Result<Vec<_>, _> = res.into_iter().map(TryInto::try_into).collect();
996 Ok(res?)
997 }
998}