agentics_persistence/db/
sessions.rs

1//! Human browser session, GitHub sign-in state, and admin service-token queries.
2
3use chrono::{DateTime, Utc};
4use sqlx::{PgPool, Postgres, Row, Transaction};
5
6use crate::db::pioneer_codes::consume_pioneer_code_for_human_tx;
7use agentics_domain::models::auth::{GithubUserId, HumanRole, HumanStatus};
8use agentics_domain::models::ids::{CreatorApiTokenId, HumanId, HumanSessionId};
9use agentics_error::{ErrorDetail, Result, ServiceError};
10
11use super::ids::{creator_api_token_id_from_row, human_id_from_row};
12
13mod account_deletion;
14mod admin_service_tokens;
15
16pub use account_deletion::{DeleteHumanAccountOutcome, delete_human_account};
17pub use admin_service_tokens::{
18    AdminServiceTokenRecord, AuthenticatedAdminServiceToken, CreateAdminServiceTokenInput,
19    authenticate_admin_service_token, create_admin_service_token, list_admin_service_tokens,
20    revoke_admin_service_token,
21};
22
23const CREATOR_API_TOKEN_ACTIVE_LABEL_INDEX: &str = "idx_creator_api_tokens_owner_active_label";
24
25/// Persisted human identity row returned to services and admin UI.
26#[derive(Debug, Clone)]
27pub struct HumanRecord {
28    pub human_id: HumanId,
29    pub status: String,
30    pub github_user_id: GithubUserId,
31    pub github_login: String,
32    pub roles: Vec<HumanRole>,
33    pub created_at: DateTime<Utc>,
34    pub disabled_at: Option<DateTime<Utc>>,
35    pub deleted_at: Option<DateTime<Utc>>,
36}
37
38/// Persisted human identity resolved from a browser session.
39#[derive(Debug, Clone)]
40pub struct AuthenticatedHumanSession {
41    pub session_id: HumanSessionId,
42    pub human_id: HumanId,
43    pub status: HumanStatus,
44    pub github_user_id: GithubUserId,
45    pub github_login: String,
46    pub roles: Vec<HumanRole>,
47    pub csrf_token_hash: String,
48    pub expires_at: DateTime<Utc>,
49}
50
51/// Persisted creator API token resolved from a bearer token.
52#[derive(Debug, Clone)]
53pub struct AuthenticatedCreatorApiToken {
54    pub token_id: CreatorApiTokenId,
55    pub human_id: HumanId,
56    pub status: HumanStatus,
57    pub github_user_id: GithubUserId,
58    pub github_login: String,
59    pub roles: Vec<HumanRole>,
60    pub label: String,
61    pub expires_at: Option<DateTime<Utc>>,
62}
63
64/// Persisted creator API-token metadata returned to the owning creator.
65#[derive(Debug, Clone)]
66pub struct CreatorApiTokenRecord {
67    pub id: CreatorApiTokenId,
68    pub label: String,
69    pub status: String,
70    pub created_by_human_id: HumanId,
71    pub created_at: DateTime<Utc>,
72    pub last_used_at: Option<DateTime<Utc>>,
73    pub expires_at: Option<DateTime<Utc>>,
74    pub revoked_at: Option<DateTime<Utc>>,
75}
76
77/// Input for resolving or creating a human from a verified GitHub identity.
78#[derive(Debug, Clone)]
79pub struct ResolveGithubHumanInput {
80    pub fallback_human_id: HumanId,
81    pub github_user_id: GithubUserId,
82    pub github_login: String,
83    pub bootstrap_admin_candidate: bool,
84}
85
86/// Input for inserting a browser session.
87#[derive(Debug, Clone)]
88pub struct CreateHumanSessionInput {
89    pub session_id: HumanSessionId,
90    pub session_token_hash: String,
91    pub csrf_token_hash: String,
92    pub human_id: HumanId,
93    pub expires_at: DateTime<Utc>,
94}
95
96/// Input for storing a short-lived GitHub sign-in state token.
97#[derive(Debug, Clone)]
98pub struct CreateGithubSignInStateInput {
99    pub state_hash: String,
100    pub browser_nonce_hash: String,
101    pub return_to: Option<String>,
102    pub expires_at: DateTime<Utc>,
103}
104
105/// Stored GitHub sign-in state consumed by a callback after browser redirect.
106#[derive(Debug, Clone)]
107pub struct ConsumedGithubSignInState {
108    pub return_to: Option<String>,
109}
110
111/// Input for creating a creator API token.
112#[derive(Debug, Clone)]
113pub struct CreateCreatorApiTokenInput {
114    pub id: CreatorApiTokenId,
115    pub token_hash: String,
116    pub label: String,
117    pub created_by_human_id: HumanId,
118    pub expires_at: Option<DateTime<Utc>>,
119}
120
121/// Resolve an existing active human or create one from a verified GitHub identity.
122pub async fn resolve_github_human(
123    pool: &PgPool,
124    input: &ResolveGithubHumanInput,
125) -> Result<HumanRecord> {
126    let mut tx = pool.begin().await?;
127
128    if let Some(existing) = find_github_human_for_update_tx(&mut tx, input.github_user_id).await? {
129        if existing.status == HumanStatus::Disabled.as_str() {
130            return Err(ServiceError::Forbidden(
131                "linked human account is disabled".to_string(),
132            ));
133        }
134        if existing.status == HumanStatus::Deleted.as_str() {
135            return Err(ServiceError::Forbidden(
136                "linked human account has been deleted".to_string(),
137            ));
138        }
139        if input.bootstrap_admin_candidate {
140            lock_bootstrap_admin_scope_tx(&mut tx).await?;
141            if !active_admin_exists_tx(&mut tx).await? {
142                activate_human_tx(&mut tx, &existing.human_id).await?;
143                grant_role_tx(&mut tx, &existing.human_id, HumanRole::Creator, None).await?;
144                grant_role_tx(&mut tx, &existing.human_id, HumanRole::Admin, None).await?;
145            }
146        }
147        sqlx::query(
148            r#"
149            UPDATE human_external_identities
150            SET provider_login = $1,
151                updated_at = NOW()
152            WHERE provider = 'github'
153              AND provider_user_id = $2
154            "#,
155        )
156        .bind(input.github_login.trim())
157        .bind(input.github_user_id.as_i64())
158        .execute(&mut *tx)
159        .await?;
160        tx.commit().await?;
161        return get_human_by_id(pool, &existing.human_id).await;
162    }
163
164    let bootstrap_admin = if input.bootstrap_admin_candidate {
165        lock_bootstrap_admin_scope_tx(&mut tx).await?;
166        !active_admin_exists_tx(&mut tx).await?
167    } else {
168        false
169    };
170
171    let initial_status = if bootstrap_admin {
172        HumanStatus::Active
173    } else {
174        HumanStatus::SetupRequired
175    };
176    insert_human_tx(&mut tx, &input.fallback_human_id, initial_status).await?;
177    insert_github_identity_tx(
178        &mut tx,
179        &input.fallback_human_id,
180        input.github_user_id,
181        &input.github_login,
182    )
183    .await?;
184
185    if bootstrap_admin {
186        grant_role_tx(&mut tx, &input.fallback_human_id, HumanRole::Creator, None).await?;
187        grant_role_tx(&mut tx, &input.fallback_human_id, HumanRole::Admin, None).await?;
188    }
189
190    tx.commit().await?;
191
192    get_human_by_id(pool, &input.fallback_human_id).await
193}
194
195/// Store a GitHub sign-in state token hash for callback validation.
196pub async fn create_github_sign_in_state(
197    pool: &PgPool,
198    input: &CreateGithubSignInStateInput,
199) -> Result<()> {
200    sqlx::query(
201        r#"
202        INSERT INTO github_sign_in_states (
203            state_hash,
204            browser_nonce_hash,
205            return_to,
206            expires_at
207        )
208        VALUES ($1, $2, $3, $4)
209        "#,
210    )
211    .bind(&input.state_hash)
212    .bind(&input.browser_nonce_hash)
213    .bind(&input.return_to)
214    .bind(input.expires_at)
215    .execute(pool)
216    .await?;
217
218    Ok(())
219}
220
221/// Consume one non-expired GitHub sign-in state token.
222pub async fn consume_github_sign_in_state(
223    pool: &PgPool,
224    state_hash: &str,
225    browser_nonce_hash: &str,
226) -> Result<Option<ConsumedGithubSignInState>> {
227    let row = sqlx::query(
228        r#"
229        DELETE FROM github_sign_in_states
230        WHERE state_hash = $1
231          AND browser_nonce_hash = $2
232          AND expires_at > NOW()
233        RETURNING return_to
234        "#,
235    )
236    .bind(state_hash)
237    .bind(browser_nonce_hash)
238    .fetch_optional(pool)
239    .await?;
240
241    let Some(row) = row else {
242        return Ok(None);
243    };
244
245    Ok(Some(ConsumedGithubSignInState {
246        return_to: row.try_get("return_to")?,
247    }))
248}
249
250/// Finish setup for one signed-in human by consuming a human pioneer code.
251pub async fn complete_human_setup(
252    pool: &PgPool,
253    human_id: &HumanId,
254    code_hash: &str,
255) -> Result<HumanRecord> {
256    let mut tx = pool.begin().await?;
257    let row = sqlx::query(
258        r#"
259        SELECT status
260        FROM humans
261        WHERE id = $1::uuid
262        FOR UPDATE
263        "#,
264    )
265    .bind(human_id.as_str())
266    .fetch_optional(&mut *tx)
267    .await?
268    .ok_or(ServiceError::NotFound)?;
269    let status: String = row.try_get("status")?;
270    if status == HumanStatus::Disabled.as_str() {
271        return Err(ServiceError::Forbidden(
272            "human account is disabled".to_string(),
273        ));
274    }
275    if status == HumanStatus::Deleted.as_str() {
276        return Err(ServiceError::Forbidden(
277            "human account has been deleted".to_string(),
278        ));
279    }
280    if status == HumanStatus::Active.as_str()
281        && active_role_exists_tx(&mut tx, human_id, HumanRole::Creator).await?
282    {
283        tx.commit().await?;
284        return get_human_by_id(pool, human_id).await;
285    }
286
287    consume_pioneer_code_for_human_tx(&mut tx, code_hash, human_id.as_str()).await?;
288    activate_human_tx(&mut tx, human_id).await?;
289    grant_role_tx(&mut tx, human_id, HumanRole::Creator, None).await?;
290    tx.commit().await?;
291
292    get_human_by_id(pool, human_id).await
293}
294
295/// Create a browser session for a verified human.
296pub async fn create_human_session(pool: &PgPool, input: &CreateHumanSessionInput) -> Result<()> {
297    sqlx::query(
298        r#"
299        INSERT INTO human_sessions (
300            id,
301            session_token_hash,
302            csrf_token_hash,
303            human_id,
304            expires_at
305        )
306        VALUES ($1::uuid, $2, $3, $4::uuid, $5)
307        "#,
308    )
309    .bind(input.session_id.as_str())
310    .bind(&input.session_token_hash)
311    .bind(&input.csrf_token_hash)
312    .bind(input.human_id.as_str())
313    .bind(input.expires_at)
314    .execute(pool)
315    .await?;
316
317    Ok(())
318}
319
320/// Authenticate a human session token and refresh its last-used timestamp.
321pub async fn authenticate_human_session(
322    pool: &PgPool,
323    session_token: &str,
324) -> Result<Option<AuthenticatedHumanSession>> {
325    let session_token_hash = crate::auth::hash_opaque_token(session_token);
326    let row = sqlx::query(
327        r#"
328        SELECT
329            s.id::text AS session_id,
330            h.id::text AS human_id,
331            h.status,
332            e.provider_user_id AS github_user_id,
333            e.provider_login AS github_login,
334            s.csrf_token_hash,
335            s.expires_at,
336            COALESCE(
337              array_agg(r.role ORDER BY r.role)
338                FILTER (WHERE r.role IS NOT NULL AND r.revoked_at IS NULL),
339              ARRAY[]::TEXT[]
340            ) AS roles
341        FROM human_sessions s
342        JOIN humans h ON h.id = s.human_id
343        JOIN human_external_identities e ON e.human_id = h.id AND e.provider = 'github'
344        LEFT JOIN human_roles r ON r.human_id = h.id
345        WHERE s.session_token_hash = $1
346          AND s.expires_at > NOW()
347          AND h.status IN ('active', 'setup_required')
348        GROUP BY s.id, h.id, h.status, e.provider_user_id, e.provider_login, s.csrf_token_hash, s.expires_at
349        LIMIT 1
350        "#,
351    )
352    .bind(&session_token_hash)
353    .fetch_optional(pool)
354    .await?;
355
356    let Some(row) = row else {
357        return Ok(None);
358    };
359
360    let session_id = HumanSessionId::try_new(row.try_get::<String, _>("session_id")?)
361        .map_err(|e| ServiceError::Internal(format!("stored invalid human session id: {e}")))?;
362    sqlx::query("UPDATE human_sessions SET last_used_at = NOW() WHERE id = $1::uuid")
363        .bind(session_id.as_str())
364        .execute(pool)
365        .await?;
366
367    Ok(Some(AuthenticatedHumanSession {
368        session_id,
369        human_id: human_id_from_row(&row, "human_id")?,
370        status: human_status_from_row(&row, "status")?,
371        github_user_id: github_user_id_from_row(&row, "github_user_id")?,
372        github_login: row.try_get("github_login")?,
373        roles: roles_from_row(&row)?,
374        csrf_token_hash: row.try_get("csrf_token_hash")?,
375        expires_at: row.try_get("expires_at")?,
376    }))
377}
378
379/// Delete a browser session by the bearer cookie token.
380pub async fn delete_human_session_by_token(pool: &PgPool, session_token: &str) -> Result<()> {
381    let session_token_hash = crate::auth::hash_opaque_token(session_token);
382    sqlx::query("DELETE FROM human_sessions WHERE session_token_hash = $1")
383        .bind(session_token_hash)
384        .execute(pool)
385        .await?;
386    Ok(())
387}
388
389/// List all human accounts for admin role management.
390pub async fn list_humans(pool: &PgPool) -> Result<Vec<HumanRecord>> {
391    let rows = sqlx::query(human_list_sql()).fetch_all(pool).await?;
392    rows.iter().map(human_record_from_row).collect()
393}
394
395/// Fetch one human by id.
396pub async fn get_human_by_id(pool: &PgPool, human_id: &HumanId) -> Result<HumanRecord> {
397    let row = sqlx::query(
398        r#"
399        SELECT
400            h.id::text AS human_id,
401            h.status,
402            h.created_at,
403            h.disabled_at,
404            h.deleted_at,
405            e.provider_user_id AS github_user_id,
406            e.provider_login AS github_login,
407            COALESCE(
408              array_agg(r.role ORDER BY r.role)
409                FILTER (WHERE r.role IS NOT NULL AND r.revoked_at IS NULL),
410              ARRAY[]::TEXT[]
411            ) AS roles
412        FROM humans h
413        JOIN human_external_identities e ON e.human_id = h.id AND e.provider = 'github'
414        LEFT JOIN human_roles r ON r.human_id = h.id
415        WHERE h.id = $1::uuid
416        GROUP BY h.id, h.status, h.created_at, h.disabled_at, h.deleted_at, e.provider_user_id, e.provider_login
417        "#,
418    )
419    .bind(human_id.as_str())
420    .fetch_optional(pool)
421    .await?
422    .ok_or(ServiceError::NotFound)?;
423    human_record_from_row(&row)
424}
425
426/// Grant the admin role to a human account.
427pub async fn grant_admin_role(
428    pool: &PgPool,
429    human_id: &HumanId,
430    granted_by_human_id: &HumanId,
431) -> Result<HumanRecord> {
432    let mut tx = pool.begin().await?;
433    lock_bootstrap_admin_scope_tx(&mut tx).await?;
434    ensure_active_human_tx(&mut tx, human_id).await?;
435    grant_role_tx(
436        &mut tx,
437        human_id,
438        HumanRole::Admin,
439        Some(granted_by_human_id),
440    )
441    .await?;
442    tx.commit().await?;
443    get_human_by_id(pool, human_id).await
444}
445
446/// Revoke the admin role from a human account.
447pub async fn revoke_admin_role(
448    pool: &PgPool,
449    human_id: &HumanId,
450    revoked_by_human_id: &HumanId,
451) -> Result<HumanRecord> {
452    let mut tx = pool.begin().await?;
453    lock_bootstrap_admin_scope_tx(&mut tx).await?;
454    let role_exists = sqlx::query(
455        r#"
456        SELECT id
457        FROM human_roles
458        WHERE human_id = $1::uuid
459          AND role = 'admin'
460          AND revoked_at IS NULL
461        FOR UPDATE
462        "#,
463    )
464    .bind(human_id.as_str())
465    .fetch_optional(&mut *tx)
466    .await?
467    .is_some();
468    if !role_exists {
469        return Err(ServiceError::NotFound);
470    }
471    let active_admin_count = active_admin_count_tx(&mut tx).await?;
472    if active_admin_count <= 1 {
473        return Err(ServiceError::BadRequest(
474            "cannot revoke the final active human admin".to_string(),
475        ));
476    }
477    sqlx::query(
478        r#"
479        UPDATE human_roles
480        SET revoked_at = COALESCE(revoked_at, NOW()),
481            revoked_by_human_id = COALESCE(revoked_by_human_id, $2::uuid)
482        WHERE human_id = $1::uuid
483          AND role = 'admin'
484          AND revoked_at IS NULL
485        "#,
486    )
487    .bind(human_id.as_str())
488    .bind(revoked_by_human_id.as_str())
489    .execute(&mut *tx)
490    .await?;
491    revoke_admin_service_tokens_created_by_human_for_admin_revocation_tx(
492        &mut tx,
493        human_id,
494        revoked_by_human_id,
495    )
496    .await?;
497    tx.commit().await?;
498    get_human_by_id(pool, human_id).await
499}
500
501async fn revoke_admin_service_tokens_created_by_human_for_admin_revocation_tx(
502    tx: &mut Transaction<'_, Postgres>,
503    human_id: &HumanId,
504    revoked_by_human_id: &HumanId,
505) -> Result<()> {
506    sqlx::query(
507        r#"
508        UPDATE admin_service_tokens
509        SET status = 'revoked',
510            revoked_at = COALESCE(revoked_at, NOW()),
511            revoked_by_human_id = COALESCE(revoked_by_human_id, $2::uuid)
512        WHERE created_by_human_id = $1::uuid
513          AND status = 'active'
514        "#,
515    )
516    .bind(human_id.as_str())
517    .bind(revoked_by_human_id.as_str())
518    .execute(&mut **tx)
519    .await?;
520    Ok(())
521}
522
523/// Create a creator API token.
524pub async fn create_creator_api_token(
525    pool: &PgPool,
526    input: &CreateCreatorApiTokenInput,
527) -> Result<CreatorApiTokenRecord> {
528    let row = sqlx::query(
529        r#"
530        INSERT INTO creator_api_tokens (
531            id,
532            token_hash,
533            label,
534            created_by_human_id,
535            expires_at
536        )
537        VALUES ($1::uuid, $2, $3, $4::uuid, $5)
538        RETURNING
539            id::text AS id,
540            label,
541            status,
542            created_by_human_id::text AS created_by_human_id,
543            created_at,
544            last_used_at,
545            expires_at,
546            revoked_at
547        "#,
548    )
549    .bind(input.id.as_str())
550    .bind(&input.token_hash)
551    .bind(input.label.trim())
552    .bind(input.created_by_human_id.as_str())
553    .bind(input.expires_at)
554    .fetch_one(pool)
555    .await
556    .map_err(map_creator_api_token_create_error)?;
557
558    creator_api_token_record_from_row(&row)
559}
560
561/// List creator API tokens owned by one human.
562pub async fn list_creator_api_tokens(
563    pool: &PgPool,
564    human_id: &HumanId,
565) -> Result<Vec<CreatorApiTokenRecord>> {
566    let rows = sqlx::query(
567        r#"
568        SELECT
569            id::text AS id,
570            label,
571            status,
572            created_by_human_id::text AS created_by_human_id,
573            created_at,
574            last_used_at,
575            expires_at,
576            revoked_at
577        FROM creator_api_tokens
578        WHERE created_by_human_id = $1::uuid
579        ORDER BY created_at DESC
580        "#,
581    )
582    .bind(human_id.as_str())
583    .fetch_all(pool)
584    .await?;
585
586    rows.iter().map(creator_api_token_record_from_row).collect()
587}
588
589/// Revoke one creator API token owned by one human.
590pub async fn revoke_creator_api_token(
591    pool: &PgPool,
592    human_id: &HumanId,
593    id: &CreatorApiTokenId,
594) -> Result<CreatorApiTokenRecord> {
595    let row = sqlx::query(
596        r#"
597        UPDATE creator_api_tokens
598        SET status = 'revoked',
599            revoked_at = COALESCE(revoked_at, NOW())
600        WHERE id = $1::uuid
601          AND created_by_human_id = $2::uuid
602        RETURNING
603            id::text AS id,
604            label,
605            status,
606            created_by_human_id::text AS created_by_human_id,
607            created_at,
608            last_used_at,
609            expires_at,
610            revoked_at
611        "#,
612    )
613    .bind(id.as_str())
614    .bind(human_id.as_str())
615    .fetch_optional(pool)
616    .await?
617    .ok_or(ServiceError::NotFound)?;
618
619    creator_api_token_record_from_row(&row)
620}
621
622/// Authenticate a creator API token by hashed bearer token.
623pub async fn authenticate_creator_api_token(
624    pool: &PgPool,
625    token_hash: &str,
626) -> Result<Option<AuthenticatedCreatorApiToken>> {
627    let row = sqlx::query(
628        r#"
629        SELECT
630            t.id::text AS id,
631            t.label,
632            t.created_by_human_id::text AS human_id,
633            t.expires_at,
634            h.status,
635            e.provider_user_id AS github_user_id,
636            e.provider_login AS github_login,
637            COALESCE(
638              array_agg(r.role ORDER BY r.role)
639                FILTER (WHERE r.role IS NOT NULL AND r.revoked_at IS NULL),
640              ARRAY[]::TEXT[]
641            ) AS roles
642        FROM creator_api_tokens t
643        JOIN humans h ON h.id = t.created_by_human_id
644        JOIN human_external_identities e ON e.human_id = h.id AND e.provider = 'github'
645        LEFT JOIN human_roles r ON r.human_id = h.id
646        WHERE t.token_hash = $1
647          AND t.status = 'active'
648          AND (t.expires_at IS NULL OR t.expires_at > NOW())
649        GROUP BY t.id, t.label, t.created_by_human_id, t.expires_at, h.status, e.provider_user_id, e.provider_login
650        LIMIT 1
651        "#,
652    )
653    .bind(token_hash)
654    .fetch_optional(pool)
655    .await?;
656
657    let Some(row) = row else {
658        return Ok(None);
659    };
660
661    let token_id = creator_api_token_id_from_row(&row, "id")?;
662    sqlx::query("UPDATE creator_api_tokens SET last_used_at = NOW() WHERE id = $1::uuid")
663        .bind(token_id.as_str())
664        .execute(pool)
665        .await?;
666
667    Ok(Some(AuthenticatedCreatorApiToken {
668        token_id,
669        label: row.try_get("label")?,
670        human_id: human_id_from_row(&row, "human_id")?,
671        status: human_status_from_row(&row, "status")?,
672        github_user_id: github_user_id_from_row(&row, "github_user_id")?,
673        github_login: row.try_get("github_login")?,
674        roles: roles_from_row(&row)?,
675        expires_at: row.try_get("expires_at")?,
676    }))
677}
678
679/// Delete expired transient auth rows.
680pub async fn delete_expired_web_auth_rows(pool: &PgPool) -> Result<()> {
681    sqlx::query("DELETE FROM github_sign_in_states WHERE expires_at <= NOW()")
682        .execute(pool)
683        .await?;
684    sqlx::query("DELETE FROM human_sessions WHERE expires_at <= NOW()")
685        .execute(pool)
686        .await?;
687    Ok(())
688}
689
690async fn find_github_human_for_update_tx(
691    tx: &mut Transaction<'_, Postgres>,
692    github_user_id: GithubUserId,
693) -> Result<Option<HumanRecord>> {
694    let row = sqlx::query(
695        r#"
696        WITH locked_human AS (
697            SELECT
698                h.id,
699                h.status,
700                h.created_at,
701                h.disabled_at,
702                h.deleted_at,
703                e.provider_user_id,
704                e.provider_login
705            FROM human_external_identities e
706            JOIN humans h ON h.id = e.human_id
707            WHERE e.provider = 'github'
708              AND e.provider_user_id = $1
709            FOR UPDATE OF h, e
710        )
711        SELECT
712            locked_human.id::text AS human_id,
713            locked_human.status,
714            locked_human.created_at,
715            locked_human.disabled_at,
716            locked_human.deleted_at,
717            locked_human.provider_user_id AS github_user_id,
718            locked_human.provider_login AS github_login,
719            COALESCE(
720              array_agg(r.role ORDER BY r.role)
721                FILTER (WHERE r.role IS NOT NULL AND r.revoked_at IS NULL),
722              ARRAY[]::TEXT[]
723            ) AS roles
724        FROM locked_human
725        LEFT JOIN human_roles r ON r.human_id = locked_human.id
726        GROUP BY
727            locked_human.id,
728            locked_human.status,
729            locked_human.created_at,
730            locked_human.disabled_at,
731            locked_human.deleted_at,
732            locked_human.provider_user_id,
733            locked_human.provider_login
734        "#,
735    )
736    .bind(github_user_id.as_i64())
737    .fetch_optional(&mut **tx)
738    .await?;
739
740    row.as_ref().map(human_record_from_row).transpose()
741}
742
743async fn insert_human_tx(
744    tx: &mut Transaction<'_, Postgres>,
745    human_id: &HumanId,
746    status: HumanStatus,
747) -> Result<()> {
748    sqlx::query(
749        r#"
750        INSERT INTO humans (id, status)
751        VALUES ($1::uuid, $2)
752        "#,
753    )
754    .bind(human_id.as_str())
755    .bind(status.as_str())
756    .execute(&mut **tx)
757    .await?;
758    Ok(())
759}
760
761async fn activate_human_tx(tx: &mut Transaction<'_, Postgres>, human_id: &HumanId) -> Result<()> {
762    sqlx::query(
763        r#"
764        UPDATE humans
765        SET status = 'active',
766            disabled_at = NULL,
767            deleted_at = NULL
768        WHERE id = $1::uuid
769          AND status NOT IN ('disabled', 'deleted')
770        "#,
771    )
772    .bind(human_id.as_str())
773    .execute(&mut **tx)
774    .await?;
775    Ok(())
776}
777
778async fn insert_github_identity_tx(
779    tx: &mut Transaction<'_, Postgres>,
780    human_id: &HumanId,
781    github_user_id: GithubUserId,
782    github_login: &str,
783) -> Result<()> {
784    sqlx::query(
785        r#"
786        INSERT INTO human_external_identities (
787            human_id,
788            provider,
789            provider_user_id,
790            provider_login
791        )
792        VALUES ($1::uuid, 'github', $2, $3)
793        "#,
794    )
795    .bind(human_id.as_str())
796    .bind(github_user_id.as_i64())
797    .bind(github_login.trim())
798    .execute(&mut **tx)
799    .await?;
800    Ok(())
801}
802
803async fn grant_role_tx(
804    tx: &mut Transaction<'_, Postgres>,
805    human_id: &HumanId,
806    role: HumanRole,
807    granted_by_human_id: Option<&HumanId>,
808) -> Result<()> {
809    sqlx::query(
810        r#"
811        INSERT INTO human_roles (id, human_id, role, granted_by_human_id)
812        VALUES ($1::uuid, $2::uuid, $3, $4::uuid)
813        ON CONFLICT (human_id, role) WHERE revoked_at IS NULL DO NOTHING
814        "#,
815    )
816    .bind(uuid::Uuid::new_v4().to_string())
817    .bind(human_id.as_str())
818    .bind(role.as_str())
819    .bind(granted_by_human_id.map(HumanId::as_str))
820    .execute(&mut **tx)
821    .await?;
822    Ok(())
823}
824
825async fn active_role_exists_tx(
826    tx: &mut Transaction<'_, Postgres>,
827    human_id: &HumanId,
828    role: HumanRole,
829) -> Result<bool> {
830    let row = sqlx::query(
831        r#"
832        SELECT id
833        FROM human_roles
834        WHERE human_id = $1::uuid
835          AND role = $2
836          AND revoked_at IS NULL
837        LIMIT 1
838        "#,
839    )
840    .bind(human_id.as_str())
841    .bind(role.as_str())
842    .fetch_optional(&mut **tx)
843    .await?;
844    Ok(row.is_some())
845}
846
847async fn ensure_active_human_tx(
848    tx: &mut Transaction<'_, Postgres>,
849    human_id: &HumanId,
850) -> Result<()> {
851    let row = sqlx::query(
852        r#"
853        SELECT status
854        FROM humans
855        WHERE id = $1::uuid
856        FOR UPDATE
857        "#,
858    )
859    .bind(human_id.as_str())
860    .fetch_optional(&mut **tx)
861    .await?
862    .ok_or(ServiceError::NotFound)?;
863    let status: String = row.try_get("status")?;
864    if status != "active" {
865        return Err(ServiceError::Forbidden(
866            "human account is not active".to_string(),
867        ));
868    }
869    Ok(())
870}
871
872async fn lock_bootstrap_admin_scope_tx(tx: &mut Transaction<'_, Postgres>) -> Result<()> {
873    sqlx::query(
874        r#"
875        INSERT INTO quota_admission_locks (scope)
876        VALUES ('global:bootstrap-admin')
877        ON CONFLICT (scope) DO NOTHING
878        "#,
879    )
880    .execute(&mut **tx)
881    .await?;
882
883    sqlx::query(
884        r#"
885        SELECT scope
886        FROM quota_admission_locks
887        WHERE scope = 'global:bootstrap-admin'
888        FOR UPDATE
889        "#,
890    )
891    .fetch_one(&mut **tx)
892    .await?;
893
894    Ok(())
895}
896
897async fn active_admin_exists_tx(tx: &mut Transaction<'_, Postgres>) -> Result<bool> {
898    Ok(active_admin_count_tx(tx).await? > 0)
899}
900
901async fn active_admin_count_tx(tx: &mut Transaction<'_, Postgres>) -> Result<i64> {
902    let row = sqlx::query(
903        r#"
904        SELECT COUNT(*) AS count
905        FROM human_roles r
906        JOIN humans h ON h.id = r.human_id
907        WHERE r.role = 'admin'
908          AND r.revoked_at IS NULL
909          AND h.status = 'active'
910        "#,
911    )
912    .fetch_one(&mut **tx)
913    .await?;
914    row.try_get("count").map_err(ServiceError::from)
915}
916
917fn human_list_sql() -> &'static str {
918    r#"
919    SELECT
920        h.id::text AS human_id,
921        h.status,
922        h.created_at,
923        h.disabled_at,
924        h.deleted_at,
925        e.provider_user_id AS github_user_id,
926        e.provider_login AS github_login,
927        COALESCE(
928          array_agg(r.role ORDER BY r.role)
929            FILTER (WHERE r.role IS NOT NULL AND r.revoked_at IS NULL),
930          ARRAY[]::TEXT[]
931        ) AS roles
932    FROM humans h
933    JOIN human_external_identities e ON e.human_id = h.id AND e.provider = 'github'
934    LEFT JOIN human_roles r ON r.human_id = h.id
935    GROUP BY h.id, h.status, h.created_at, h.disabled_at, h.deleted_at, e.provider_user_id, e.provider_login
936    ORDER BY h.created_at DESC
937    "#
938}
939
940fn human_record_from_row(row: &sqlx::postgres::PgRow) -> Result<HumanRecord> {
941    Ok(HumanRecord {
942        human_id: human_id_from_row(row, "human_id")?,
943        status: row.try_get("status")?,
944        github_user_id: github_user_id_from_row(row, "github_user_id")?,
945        github_login: row.try_get("github_login")?,
946        roles: roles_from_row(row)?,
947        created_at: row.try_get("created_at")?,
948        disabled_at: row.try_get("disabled_at")?,
949        deleted_at: row.try_get("deleted_at")?,
950    })
951}
952
953fn github_user_id_from_row(row: &sqlx::postgres::PgRow, field: &str) -> Result<GithubUserId> {
954    let value = row.try_get::<i64, _>(field)?;
955    GithubUserId::try_new(value)
956        .map_err(|e| ServiceError::Internal(format!("stored invalid GitHub user id: {e}")))
957}
958
959fn human_status_from_row(row: &sqlx::postgres::PgRow, field: &str) -> Result<HumanStatus> {
960    let value = row.try_get::<String, _>(field)?;
961    HumanStatus::from_storage_value(&value)
962        .ok_or_else(|| ServiceError::Internal(format!("stored invalid human status `{value}`")))
963}
964
965fn roles_from_row(row: &sqlx::postgres::PgRow) -> Result<Vec<HumanRole>> {
966    let roles = row.try_get::<Vec<String>, _>("roles")?;
967    roles
968        .into_iter()
969        .map(|role| {
970            HumanRole::from_storage_value(&role).ok_or_else(|| {
971                ServiceError::Internal(format!("stored invalid human role `{role}`"))
972            })
973        })
974        .collect()
975}
976
977fn creator_api_token_record_from_row(row: &sqlx::postgres::PgRow) -> Result<CreatorApiTokenRecord> {
978    Ok(CreatorApiTokenRecord {
979        id: creator_api_token_id_from_row(row, "id")?,
980        label: row.try_get("label")?,
981        status: row.try_get("status")?,
982        created_by_human_id: human_id_from_row(row, "created_by_human_id")?,
983        created_at: row.try_get("created_at")?,
984        last_used_at: row.try_get("last_used_at")?,
985        expires_at: row.try_get("expires_at")?,
986        revoked_at: row.try_get("revoked_at")?,
987    })
988}
989
990fn map_creator_api_token_create_error(error: sqlx::Error) -> ServiceError {
991    match error {
992        sqlx::Error::Database(db_err)
993            if db_err.is_unique_violation()
994                && db_err
995                    .constraint()
996                    .is_some_and(|name| name == CREATOR_API_TOKEN_ACTIVE_LABEL_INDEX) =>
997        {
998            duplicate_token_label_conflict(
999                "active creator API token label already exists for this human",
1000                "An active creator API token already uses this label.",
1001            )
1002        }
1003        error => ServiceError::Database(error),
1004    }
1005}
1006
1007fn duplicate_token_label_conflict(message: &str, detail_message: &str) -> ServiceError {
1008    ServiceError::conflict_with_details(
1009        message,
1010        [ErrorDetail {
1011            field: Some("label".to_string()),
1012            message: detail_message.to_string(),
1013        }],
1014    )
1015}
agentics_persistence/db/sessions.rs

agentics_persistence/db/
sessions.rs
