pharos_runtime_database/

account_tokens.rs

//! Atomic password-reset completion over the configured database adapter.

use std::collections::BTreeMap;

use crate::DbBackendConfig;
use crate::operations::{
    AppSchema, DatabaseError, DatabaseOperations, Mutation, Predicate, Query, Transaction,
    WriteValue,
};
use serde_json::Value;

/// Reads the selected account and profile for app-owned profile rendering.
/// Credentials and other users never enter this view.
pub fn account_profile_view(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    actor_id: i64,
) -> Result<Value, DatabaseError> {
    if actor_id <= 0 {
        return Err(DatabaseError::AccessDenied);
    }
    let mut accounts = adapter.query(
        config,
        schema,
        &Query {
            entity: "user_account".to_owned(),
            fields: ["id", "username", "email", "display_name", "is_active"]
                .into_iter()
                .map(ToOwned::to_owned)
                .collect(),
            filter: Some(Predicate::All(vec![
                Predicate::Equal("id".to_owned(), Value::from(actor_id)),
                Predicate::Equal("is_active".to_owned(), Value::Bool(true)),
            ])),
            order: Vec::new(),
            limit: 1,
            offset: 0,
        },
    )?;
    let mut account = accounts.pop().ok_or(DatabaseError::MissingRecord)?;
    let profiles = adapter.query(
        config,
        schema,
        &Query {
            entity: "user_profile".to_owned(),
            fields: schema
                .entities
                .iter()
                .find(|entity| entity.name == "user_profile")
                .ok_or(DatabaseError::InvalidContract)?
                .fields
                .iter()
                .map(|field| field.name.clone())
                .collect(),
            filter: Some(Predicate::Equal(
                "user_id".to_owned(),
                Value::from(actor_id),
            )),
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if profiles.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    let profile = profiles
        .into_iter()
        .next()
        .ok_or(DatabaseError::MissingRecord)?;
    account.extend(
        profile
            .into_iter()
            .filter(|(name, _)| !matches!(name.as_str(), "id" | "user_id" | "created_at")),
    );
    let subscriptions = adapter.query(
        config,
        schema,
        &Query {
            entity: "subscription".to_owned(),
            fields: vec![
                "id".to_owned(),
                "user_id".to_owned(),
                "business_id".to_owned(),
            ],
            filter: Some(Predicate::Equal(
                "user_id".to_owned(),
                Value::from(actor_id),
            )),
            order: Vec::new(),
            limit: 257,
            offset: 0,
        },
    )?;
    if subscriptions.len() > 256 {
        return Err(DatabaseError::CapacityExceeded);
    }
    let ids = subscriptions
        .iter()
        .map(|row| {
            row.get("business_id")
                .cloned()
                .ok_or(DatabaseError::StorageFailure)
        })
        .collect::<Result<Vec<_>, _>>()?;
    let businesses = crate::operations::query_related_rows(
        adapter,
        config,
        schema,
        crate::operations::RelatedRowsQuery {
            entity: "business",
            foreign_key: "id",
            keys: &ids,
            fields: &["id".to_owned(), "name".to_owned(), "slug".to_owned()],
            maximum: 256,
        },
    )?;
    if businesses.len() > 256 {
        return Err(DatabaseError::CapacityExceeded);
    }
    let mut calendars = adapter.query(
        config,
        schema,
        &Query {
            entity: "google_calendar_connection".to_owned(),
            fields: vec!["user_id".to_owned(), "status".to_owned(), "mode".to_owned()],
            filter: Some(Predicate::All(vec![
                Predicate::Equal("user_id".to_owned(), Value::from(actor_id)),
                Predicate::Equal("mode".to_owned(), Value::String("user".to_owned())),
            ])),
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if calendars.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    for calendar in &mut calendars {
        let status = calendar
            .get("status")
            .cloned()
            .ok_or(DatabaseError::StorageFailure)?;
        calendar.insert("connection_status".to_owned(), status);
    }
    Ok(
        serde_json::json!({"users": [account], "subscriptions": subscriptions, "businesses": businesses, "google_calendar_connections": calendars}),
    )
}

/// Updates only the signed-in actor's editable profile fields. Privileged
/// account type, verification state, credentials, and memberships are excluded.
pub fn update_account_profile(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    actor_id: i64,
    input: &serde_json::Map<String, Value>,
) -> Result<(), DatabaseError> {
    if actor_id <= 0
        || input.is_empty()
        || input.keys().any(|name| {
            !matches!(
                name.as_str(),
                "full_name" | "phone_number" | "age" | "address" | "payment_preference"
            )
        })
        || input
            .get("full_name")
            .and_then(Value::as_str)
            .is_none_or(|value| value.trim().is_empty() || value.len() > 256)
    {
        return Err(DatabaseError::InvalidInput);
    }
    if input
        .get("phone_number")
        .is_some_and(|value| value.as_str().is_none_or(|value| value.len() > 64))
        || input
            .get("address")
            .is_some_and(|value| value.as_str().is_none_or(|value| value.len() > 2048))
        || input.get("age").is_some_and(|value| {
            !value.is_null() && value.as_i64().is_none_or(|age| !(0..=150).contains(&age))
        })
        || input.get("payment_preference").is_some_and(|value| {
            value
                .as_str()
                .is_none_or(|value| !matches!(value, "pay_on_site" | "credit_card"))
        })
    {
        return Err(DatabaseError::InvalidInput);
    }
    let ownership = Predicate::Equal("user_id".to_owned(), Value::from(actor_id));
    let rows = adapter.query(
        config,
        schema,
        &Query {
            entity: "user_profile".to_owned(),
            fields: input.keys().cloned().collect(),
            filter: Some(ownership.clone()),
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if rows.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    let current = rows.first().ok_or(DatabaseError::MissingRecord)?;
    let mut guards = vec![ownership];
    let mut values = BTreeMap::new();
    for (field, value) in input {
        let previous = current.get(field).ok_or(DatabaseError::StorageFailure)?;
        if previous == value {
            continue;
        }
        guards.push(if previous.is_null() {
            Predicate::IsNull(field.clone())
        } else {
            Predicate::Equal(field.clone(), previous.clone())
        });
        values.insert(field.clone(), WriteValue::Literal(value.clone()));
    }
    if values.is_empty() {
        return Ok(());
    }
    adapter.transact(
        config,
        schema,
        &Transaction {
            steps: vec![
                Mutation::RequireCount {
                    entity: "user_account".to_owned(),
                    filter: Predicate::All(vec![
                        Predicate::Equal("id".to_owned(), Value::from(actor_id)),
                        Predicate::Equal("is_active".to_owned(), Value::Bool(true)),
                    ]),
                    minimum: 1,
                    maximum: Some(1),
                },
                Mutation::Update {
                    entity: "user_profile".to_owned(),
                    filter: Predicate::All(guards),
                    values,
                    expected_rows: Some(1),
                },
            ],
            idempotency_key: None,
        },
    )?;
    Ok(())
}

/// Creates a customer account and its profile in one configured-backend transaction.
pub fn register_account(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    username: &str,
    email: &str,
    password_hash: &str,
) -> Result<i64, DatabaseError> {
    if username.is_empty()
        || username.len() > 128
        || !username
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
        || email.is_empty()
        || email.len() > 254
        || email.split_once('@').is_none_or(|(local, domain)| {
            local.is_empty() || domain.is_empty() || !domain.contains('.')
        })
        || password_hash.is_empty()
        || password_hash.len() > 1024
    {
        return Err(DatabaseError::InvalidInput);
    }
    let result = adapter.transact(
        config,
        schema,
        &Transaction {
            steps: vec![
                Mutation::Insert {
                    entity: "user_account".to_owned(),
                    values: BTreeMap::from([
                        (
                            "username".to_owned(),
                            WriteValue::Literal(Value::String(username.to_owned())),
                        ),
                        (
                            "email".to_owned(),
                            WriteValue::Literal(Value::String(email.to_owned())),
                        ),
                        (
                            "password_hash".to_owned(),
                            WriteValue::Literal(Value::String(password_hash.to_owned())),
                        ),
                        (
                            "is_active".to_owned(),
                            WriteValue::Literal(Value::Bool(true)),
                        ),
                    ]),
                },
                Mutation::Insert {
                    entity: "user_profile".to_owned(),
                    values: BTreeMap::from([
                        ("user_id".to_owned(), WriteValue::GeneratedId(0)),
                        (
                            "account_type".to_owned(),
                            WriteValue::Literal(Value::String("business_customer".to_owned())),
                        ),
                        (
                            "payment_preference".to_owned(),
                            WriteValue::Literal(Value::String("pay_on_site".to_owned())),
                        ),
                        (
                            "email_verified".to_owned(),
                            WriteValue::Literal(Value::Bool(false)),
                        ),
                    ]),
                },
            ],
            idempotency_key: None,
        },
    )?;
    result
        .generated_ids
        .first()
        .copied()
        .flatten()
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)
}

/// Persists one reset token and its delivery work in a single adapter transaction.
pub fn issue_password_reset(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    email: &str,
    token_hash: &str,
    delivery_secret: &str,
    expires_at: &str,
) -> Result<bool, DatabaseError> {
    if email.len() > 320
        || email.split_once('@').is_none_or(|(local, domain)| {
            local.is_empty() || domain.is_empty() || !domain.contains('.')
        })
        || token_hash.len() != 64
        || !token_hash.bytes().all(|byte| byte.is_ascii_hexdigit())
        || delivery_secret.len() != 64
        || !delivery_secret.bytes().all(|byte| byte.is_ascii_hexdigit())
        || expires_at.is_empty()
        || expires_at.len() > 64
    {
        return Err(DatabaseError::InvalidInput);
    }
    let users = adapter.query(
        config,
        schema,
        &Query {
            entity: "user_account".to_owned(),
            fields: vec!["id".to_owned()],
            filter: Some(Predicate::Equal(
                "email".to_owned(),
                Value::String(email.to_owned()),
            )),
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if users.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    let Some(user) = users.first() else {
        return Ok(false);
    };
    let user_id = user
        .get("id")
        .and_then(Value::as_i64)
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)?;
    adapter.transact(
        config,
        schema,
        &Transaction {
            steps: vec![
                Mutation::RequireCount {
                    entity: "user_account".to_owned(),
                    filter: Predicate::All(vec![
                        Predicate::Equal("id".to_owned(), Value::from(user_id)),
                        Predicate::Equal("email".to_owned(), Value::String(email.to_owned())),
                    ]),
                    minimum: 1,
                    maximum: Some(1),
                },
                Mutation::Insert {
                    entity: "password_reset".to_owned(),
                    values: BTreeMap::from([
                        (
                            "user_id".to_owned(),
                            WriteValue::Literal(Value::from(user_id)),
                        ),
                        (
                            "token_hash".to_owned(),
                            WriteValue::Literal(Value::String(token_hash.to_owned())),
                        ),
                        (
                            "expires_at".to_owned(),
                            WriteValue::Literal(Value::String(expires_at.to_owned())),
                        ),
                    ]),
                },
                Mutation::Insert {
                    entity: "notification_delivery".to_owned(),
                    values: BTreeMap::from([
                        (
                            "adapter_id".to_owned(),
                            WriteValue::Literal(Value::String("local_reset_queue".to_owned())),
                        ),
                        (
                            "channel".to_owned(),
                            WriteValue::Literal(Value::String("email".to_owned())),
                        ),
                        (
                            "recipient".to_owned(),
                            WriteValue::Literal(Value::String(email.to_owned())),
                        ),
                        (
                            "template_id".to_owned(),
                            WriteValue::Literal(Value::String("password_reset".to_owned())),
                        ),
                        (
                            "delivery_status".to_owned(),
                            WriteValue::Literal(Value::String("queued".to_owned())),
                        ),
                        (
                            "resource_id".to_owned(),
                            WriteValue::Literal(Value::String(delivery_secret.to_owned())),
                        ),
                        (
                            "max_attempts".to_owned(),
                            WriteValue::Literal(Value::from(0)),
                        ),
                        (
                            "backoff_seconds".to_owned(),
                            WriteValue::Literal(Value::from(0)),
                        ),
                    ]),
                },
            ],
            idempotency_key: None,
        },
    )?;
    Ok(true)
}

/// Claims a live reset token, changes its account password, and revokes sessions atomically.
pub fn complete_password_reset(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    token_hash: &str,
    password_hash: &str,
    now: &str,
) -> Result<Option<i64>, DatabaseError> {
    if token_hash.len() != 64
        || !token_hash.bytes().all(|byte| byte.is_ascii_hexdigit())
        || password_hash.is_empty()
        || password_hash.len() > 1024
        || now.is_empty()
        || now.len() > 64
    {
        return Err(DatabaseError::InvalidInput);
    }
    let live_token = Predicate::All(vec![
        Predicate::Equal(
            "token_hash".to_owned(),
            Value::String(token_hash.to_owned()),
        ),
        Predicate::IsNull("claimed_at".to_owned()),
        Predicate::GreaterThan("expires_at".to_owned(), Value::String(now.to_owned())),
    ]);
    let resets = adapter.query(
        config,
        schema,
        &Query {
            entity: "password_reset".to_owned(),
            fields: vec!["id".to_owned(), "user_id".to_owned()],
            filter: Some(live_token.clone()),
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if resets.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    let Some(reset) = resets.first() else {
        return Ok(None);
    };
    let reset_id = reset
        .get("id")
        .and_then(Value::as_i64)
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)?;
    let user_id = reset
        .get("user_id")
        .and_then(Value::as_i64)
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)?;
    let reset_filter = Predicate::All(vec![
        live_token,
        Predicate::Equal("id".to_owned(), Value::from(reset_id)),
    ]);
    let transaction = Transaction {
        steps: vec![
            Mutation::RequireCount {
                entity: "password_reset".to_owned(),
                filter: reset_filter.clone(),
                minimum: 1,
                maximum: Some(1),
            },
            Mutation::Update {
                entity: "password_reset".to_owned(),
                filter: reset_filter,
                values: BTreeMap::from([(
                    "claimed_at".to_owned(),
                    WriteValue::Literal(Value::String(now.to_owned())),
                )]),
                expected_rows: Some(1),
            },
            Mutation::Update {
                entity: "user_account".to_owned(),
                filter: Predicate::Equal("id".to_owned(), Value::from(user_id)),
                values: BTreeMap::from([(
                    "password_hash".to_owned(),
                    WriteValue::Literal(Value::String(password_hash.to_owned())),
                )]),
                expected_rows: Some(1),
            },
            Mutation::Update {
                entity: "auth_session".to_owned(),
                filter: Predicate::All(vec![
                    Predicate::Equal("user_id".to_owned(), Value::from(user_id)),
                    Predicate::IsNull("revoked_at".to_owned()),
                ]),
                values: BTreeMap::from([(
                    "revoked_at".to_owned(),
                    WriteValue::Literal(Value::String(now.to_owned())),
                )]),
                expected_rows: None,
            },
        ],
        idempotency_key: None,
    };
    adapter.transact(config, schema, &transaction)?;
    Ok(Some(user_id))
}

/// Claims a live email token and marks its account profile verified in one transaction.
pub fn complete_email_verification(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    token_hash: &str,
    now: &str,
) -> Result<Option<i64>, DatabaseError> {
    if token_hash.is_empty()
        || token_hash.len() > 256
        || token_hash.contains(['\0', '\r', '\n'])
        || now.is_empty()
        || now.len() > 64
    {
        return Err(DatabaseError::InvalidInput);
    }
    let live_token = Predicate::All(vec![
        Predicate::Equal(
            "token_hash".to_owned(),
            Value::String(token_hash.to_owned()),
        ),
        Predicate::IsNull("claimed_at".to_owned()),
        Predicate::GreaterThan("expires_at".to_owned(), Value::String(now.to_owned())),
    ]);
    let verifications = adapter.query(
        config,
        schema,
        &Query {
            entity: "email_verification".to_owned(),
            fields: vec!["id".to_owned(), "user_id".to_owned()],
            filter: Some(live_token.clone()),
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if verifications.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    let Some(verification) = verifications.first() else {
        return Ok(None);
    };
    let verification_id = verification
        .get("id")
        .and_then(Value::as_i64)
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)?;
    let user_id = verification
        .get("user_id")
        .and_then(Value::as_i64)
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)?;
    let filter = Predicate::All(vec![
        live_token,
        Predicate::Equal("id".to_owned(), Value::from(verification_id)),
    ]);
    adapter.transact(
        config,
        schema,
        &Transaction {
            steps: vec![
                Mutation::RequireCount {
                    entity: "email_verification".to_owned(),
                    filter: filter.clone(),
                    minimum: 1,
                    maximum: Some(1),
                },
                Mutation::Update {
                    entity: "email_verification".to_owned(),
                    filter,
                    values: BTreeMap::from([(
                        "claimed_at".to_owned(),
                        WriteValue::Literal(Value::String(now.to_owned())),
                    )]),
                    expected_rows: Some(1),
                },
                Mutation::Update {
                    entity: "user_profile".to_owned(),
                    filter: Predicate::Equal("user_id".to_owned(), Value::from(user_id)),
                    values: BTreeMap::from([(
                        "email_verified".to_owned(),
                        WriteValue::Literal(Value::Bool(true)),
                    )]),
                    expected_rows: Some(1),
                },
            ],
            idempotency_key: None,
        },
    )?;
    Ok(Some(user_id))
}