pharos_runtime_database/

payment.rs

//! Bounded payment settings reads through the configured app database.

use std::collections::BTreeMap;

use serde_json::{Map, Value};

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

fn declared_text<'a>(declaration: &'a Value, key: &str) -> Result<&'a str, DatabaseError> {
    declaration
        .get(key)
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty() && value.len() <= 128)
        .ok_or(DatabaseError::InvalidContract)
}

fn declared_fields(
    schema: &AppSchema,
    entity: &str,
    names: &[&str],
) -> Result<Vec<String>, DatabaseError> {
    let declared = schema
        .entities
        .iter()
        .find(|candidate| candidate.name == entity)
        .ok_or(DatabaseError::InvalidContract)?;
    let mut fields = Vec::new();
    for name in names {
        if !declared.fields.iter().any(|field| field.name == *name) {
            return Err(DatabaseError::InvalidContract);
        }
        if !fields.iter().any(|field| field == name) {
            fields.push((*name).to_owned());
        }
    }
    Ok(fields)
}

/// Selects the full declared row only for internal lifecycle processing.
fn all_fields(schema: &AppSchema, entity: &str) -> Result<Vec<String>, DatabaseError> {
    Ok(schema
        .entities
        .iter()
        .find(|definition| definition.name == entity)
        .ok_or(DatabaseError::InvalidContract)?
        .fields
        .iter()
        .map(|field| field.name.clone())
        .collect())
}

fn one_row(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    entity: &str,
    fields: Vec<String>,
    filter: Option<Predicate>,
) -> Result<Option<Map<String, Value>>, DatabaseError> {
    let mut rows = adapter.query(
        config,
        schema,
        &Query {
            entity: entity.to_owned(),
            fields,
            filter,
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if rows.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    Ok(rows.pop())
}

/// Reads only the owner and payment records needed to render provider settings.
/// The caller supplies an app-owned owner declaration; no backend or app id is
/// selected here. Secret ciphertext is replaced with a presence marker before
/// the returned view leaves this module.
pub fn payment_render_state(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    declaration: &Value,
    owner_route_value: Option<&str>,
    provider_id: &str,
) -> Result<Value, DatabaseError> {
    payment_state(
        adapter,
        config,
        schema,
        declaration,
        owner_route_value,
        provider_id,
        None,
    )
}

/// Loads a bounded transaction view for one owner and authenticated actor.
pub fn payment_lifecycle_state(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    declaration: &Value,
    owner_route_value: &str,
    provider_id: &str,
    actor_id: i64,
) -> Result<Value, DatabaseError> {
    if actor_id <= 0 {
        return Err(DatabaseError::InvalidInput);
    }
    payment_state(
        adapter,
        config,
        schema,
        declaration,
        Some(owner_route_value),
        provider_id,
        Some(actor_id),
    )
}

fn payment_state(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    declaration: &Value,
    owner_route_value: Option<&str>,
    provider_id: &str,
    actor_id: Option<i64>,
) -> Result<Value, DatabaseError> {
    if provider_id.is_empty() || provider_id.len() > 64 {
        return Err(DatabaseError::InvalidInput);
    }
    let owner_entity = declared_text(declaration, "owner_entity")?;
    let owner_section = declared_text(declaration, "owner_state_section")?;
    let owner_route_field = declared_text(declaration, "owner_route_field")?;
    let owner_id_field = declared_text(declaration, "owner_id_field")?;
    let owner_type = declared_text(declaration, "owner_resource_type")?;
    let tenant_entity = declared_text(declaration, "tenant_entity")?;
    let tenant_section = declared_text(declaration, "tenant_state_section")?;
    let tenant_owner_id_field = declared_text(declaration, "tenant_owner_id_field")?;
    let tenant_id_field = declared_text(declaration, "tenant_id_field")?;
    let mut owner_names = vec![owner_route_field, owner_id_field];
    if let Some(field) = declaration
        .get("owner_user_id_field")
        .and_then(Value::as_str)
    {
        owner_names.push(field);
    }
    if tenant_entity == owner_entity {
        owner_names.push(tenant_owner_id_field);
        owner_names.push(tenant_id_field);
        if let Some(field) = declaration
            .get("tenant_owner_user_id_field")
            .and_then(Value::as_str)
        {
            owner_names.push(field);
        }
    }
    let owner_fields = declared_fields(schema, owner_entity, &owner_names)?;
    let owner_filter = owner_route_value
        .map(|value| {
            if value.is_empty() || value.len() > 128 {
                return Err(DatabaseError::InvalidInput);
            }
            Ok(Predicate::Equal(
                owner_route_field.to_owned(),
                Value::String(value.to_owned()),
            ))
        })
        .transpose()?;
    let owner_result = one_row(
        adapter,
        config,
        schema,
        owner_entity,
        owner_fields,
        owner_filter,
    );
    if owner_route_value.is_none() && matches!(owner_result, Err(DatabaseError::Conflict)) {
        return Ok(Value::Object(Map::new()));
    }
    let Some(owner) = owner_result? else {
        return Ok(Value::Object(Map::new()));
    };
    let owner_id = owner
        .get(owner_id_field)
        .and_then(Value::as_i64)
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)?;
    let tenant = if tenant_entity == owner_entity && tenant_section == owner_section {
        owner.clone()
    } else {
        let mut tenant_names = vec![tenant_id_field, tenant_owner_id_field];
        if let Some(field) = declaration
            .get("tenant_owner_user_id_field")
            .and_then(Value::as_str)
        {
            tenant_names.push(field);
        }
        let tenant_fields = declared_fields(schema, tenant_entity, &tenant_names)?;
        let Some(row) = one_row(
            adapter,
            config,
            schema,
            tenant_entity,
            tenant_fields,
            Some(Predicate::Equal(
                tenant_owner_id_field.to_owned(),
                Value::from(owner_id),
            )),
        )?
        else {
            return Ok(Value::Object(Map::new()));
        };
        row
    };
    let tenant_id = tenant
        .get(tenant_id_field)
        .and_then(Value::as_i64)
        .filter(|id| *id > 0)
        .ok_or(DatabaseError::StorageFailure)?;
    let scope = Predicate::All(vec![
        Predicate::Equal("app_id".to_owned(), Value::String(schema.app_id.clone())),
        Predicate::Equal("tenant_id".to_owned(), Value::from(tenant_id)),
        Predicate::Equal(
            "owner_resource_type".to_owned(),
            Value::String(owner_type.to_owned()),
        ),
        Predicate::Equal("owner_resource_id".to_owned(), Value::from(owner_id)),
        Predicate::Equal(
            "provider_id".to_owned(),
            Value::String(provider_id.to_owned()),
        ),
    ]);
    let account_fields = if actor_id.is_some() {
        all_fields(schema, "payment_provider_account")?
    } else {
        declared_fields(
            schema,
            "payment_provider_account",
            &[
                "app_id",
                "id",
                "tenant_id",
                "owner_resource_type",
                "owner_resource_id",
                "provider_id",
                "checkout_mode",
                "connection_state",
                "capability_state",
                "onboarding_state",
                "provider_account_id",
                "provider_account_label",
                "status_reason_code",
                "status_reason_detail",
                "charges_enabled",
                "payouts_enabled",
                "details_submitted",
            ],
        )?
    };
    let accounts = adapter.query(
        config,
        schema,
        &Query {
            entity: "payment_provider_account".to_owned(),
            fields: account_fields,
            filter: Some(scope.clone()),
            order: Vec::new(),
            limit: 2,
            offset: 0,
        },
    )?;
    if accounts.len() > 1 {
        return Err(DatabaseError::Conflict);
    }
    let secret_fields = if actor_id.is_some() {
        all_fields(schema, "payment_provider_secret")?
    } else {
        declared_fields(
            schema,
            "payment_provider_secret",
            &[
                "app_id",
                "tenant_id",
                "owner_resource_type",
                "owner_resource_id",
                "provider_id",
                "secret_kind",
                "ciphertext",
                "revoked_at",
            ],
        )?
    };
    let mut secrets = adapter.query(
        config,
        schema,
        &Query {
            entity: "payment_provider_secret".to_owned(),
            fields: secret_fields,
            filter: Some(scope.clone()),
            order: Vec::new(),
            limit: 17,
            offset: 0,
        },
    )?;
    if secrets.len() > 16 {
        return Err(DatabaseError::CapacityExceeded);
    }
    for row in secrets.iter_mut().filter(|_| actor_id.is_none()) {
        let present = row
            .get("ciphertext")
            .and_then(Value::as_str)
            .is_some_and(|text| !text.is_empty());
        row.insert(
            "ciphertext".to_owned(),
            Value::String(if present { "present" } else { "" }.to_owned()),
        );
    }
    let mut root = Map::new();
    root.insert(
        owner_section.to_owned(),
        Value::Array(vec![Value::Object(owner)]),
    );
    if tenant_section != owner_section {
        root.insert(
            tenant_section.to_owned(),
            Value::Array(vec![Value::Object(tenant)]),
        );
    }
    root.insert(
        "payment_provider_accounts".to_owned(),
        Value::Array(accounts.into_iter().map(Value::Object).collect()),
    );
    root.insert(
        "payment_provider_secrets".to_owned(),
        Value::Array(secrets.into_iter().map(Value::Object).collect()),
    );
    if let Some(actor_id) = actor_id {
        root.insert(
            "_payment_scope".to_owned(),
            serde_json::json!({
                "app_id": schema.app_id, "tenant_id": tenant_id,
                "owner_resource_type": owner_type, "owner_resource_id": owner_id,
                "provider_id": provider_id, "actor_user_id": actor_id,
            }),
        );
        if let Some(entity) = declaration.get("membership_entity").and_then(Value::as_str) {
            let section = declared_text(declaration, "membership_state_section")?;
            let user_field = declared_text(declaration, "membership_user_id_field")?;
            let tenant_field = declared_text(declaration, "membership_tenant_id_field")?;
            let status_field = declared_text(declaration, "membership_status_field")?;
            let mut membership_fields = vec![user_field, tenant_field, status_field];
            for key in ["membership_role_field", "membership_revoked_field"] {
                if let Some(field) = declaration.get(key).and_then(Value::as_str) {
                    membership_fields.push(field);
                }
            }
            let rows = adapter.query(
                config,
                schema,
                &Query {
                    entity: entity.to_owned(),
                    fields: declared_fields(schema, entity, &membership_fields)?,
                    filter: Some(Predicate::All(vec![
                        Predicate::Equal(user_field.to_owned(), Value::from(actor_id)),
                        Predicate::Equal(tenant_field.to_owned(), Value::from(tenant_id)),
                    ])),
                    order: Vec::new(),
                    limit: 2,
                    offset: 0,
                },
            )?;
            if rows.len() > 1 {
                return Err(DatabaseError::Conflict);
            }
            root.insert(
                section.to_owned(),
                Value::Array(rows.into_iter().map(Value::Object).collect()),
            );
        }
        let epoch = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_err(|_| DatabaseError::StorageFailure)?
            .as_secs();
        let now = i64::try_from(epoch)
            .ok()
            .and_then(|epoch| chrono::DateTime::<chrono::Utc>::from_timestamp(epoch, 0))
            .ok_or(DatabaseError::StorageFailure)?
            .to_rfc3339();
        let nonces = adapter.query(
            config,
            schema,
            &Query {
                entity: "payment_provider_callback_nonce".to_owned(),
                fields: all_fields(schema, "payment_provider_callback_nonce")?,
                filter: Some(Predicate::All(vec![
                    scope,
                    Predicate::IsNull("claimed_at".to_owned()),
                    Predicate::GreaterThan("expires_at".to_owned(), Value::String(now)),
                ])),
                order: Vec::new(),
                limit: 257,
                offset: 0,
            },
        )?;
        if nonces.len() > 256 {
            return Err(DatabaseError::CapacityExceeded);
        }
        root.insert(
            "payment_provider_callback_nonces".to_owned(),
            Value::Array(nonces.into_iter().map(Value::Object).collect()),
        );
        root.insert(
            "payment_provider_audit_events".to_owned(),
            Value::Array(Vec::new()),
        );
        // Legacy service constructors allocate provisional IDs from their view.
        // A single indexed maximum per entity supplies the seed; the transaction
        // below replaces provisional IDs with backend-generated identities.
        for (section, entity) in PAYMENT_SECTIONS {
            let latest = adapter.query(
                config,
                schema,
                &Query {
                    entity: (*entity).to_owned(),
                    fields: vec!["id".to_owned()],
                    filter: None,
                    order: vec![SortKey {
                        field: "id".to_owned(),
                        descending: true,
                    }],
                    limit: 1,
                    offset: 0,
                },
            )?;
            let rows = root
                .get_mut(*section)
                .and_then(Value::as_array_mut)
                .ok_or(DatabaseError::InvalidInput)?;
            for row in latest {
                if !rows
                    .iter()
                    .any(|existing| existing.get("id") == row.get("id"))
                {
                    rows.push(Value::Object(row));
                }
            }
        }
    }
    Ok(Value::Object(root))
}

const PAYMENT_SECTIONS: &[(&str, &str)] = &[
    ("payment_provider_secrets", "payment_provider_secret"),
    ("payment_provider_accounts", "payment_provider_account"),
    (
        "payment_provider_callback_nonces",
        "payment_provider_callback_nonce",
    ),
    (
        "payment_provider_audit_events",
        "payment_provider_audit_event",
    ),
];

fn identity_reference<'a>(entity: &str, field: &'a crate::operations::Field) -> Option<&'a str> {
    if entity == "payment_provider_account" && field.name == "secret_reference_id" {
        Some("payment_provider_secret")
    } else {
        field
            .references
            .as_ref()
            .filter(|(_, target)| target == "id")
            .map(|(target, _)| target.as_str())
    }
}

/// Commits changed payment rows atomically, with compare-and-write guards and
/// transaction-local references for identities generated by any backend.
pub fn apply_payment_lifecycle_delta(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    declaration: &Value,
    actor_id: i64,
    before: &Value,
    after: &Value,
) -> Result<Value, DatabaseError> {
    let scope = before
        .get("_payment_scope")
        .and_then(Value::as_object)
        .ok_or(DatabaseError::InvalidInput)?;
    if scope.get("app_id").and_then(Value::as_str) != Some(schema.app_id.as_str())
        || scope.get("actor_user_id").and_then(Value::as_i64) != Some(actor_id)
        || before.get("_payment_scope") != after.get("_payment_scope")
    {
        return Err(DatabaseError::InvalidInput);
    }
    let mut 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),
        },
        payment_authorization_guard(declaration, actor_id, before)?,
    ];
    let mut inserted = BTreeMap::new();
    for &(section, entity_name) in PAYMENT_SECTIONS {
        let entity = schema
            .entities
            .iter()
            .find(|entity| entity.name == entity_name)
            .ok_or(DatabaseError::InvalidContract)?;
        let old_rows = before
            .get(section)
            .and_then(Value::as_array)
            .ok_or(DatabaseError::InvalidInput)?;
        let new_rows = after
            .get(section)
            .and_then(Value::as_array)
            .ok_or(DatabaseError::InvalidInput)?;
        let mut old_by_id = BTreeMap::new();
        for row in old_rows {
            let id = row
                .get("id")
                .and_then(Value::as_i64)
                .filter(|id| *id > 0)
                .ok_or(DatabaseError::InvalidInput)?;
            if old_by_id.insert(id, row).is_some() {
                return Err(DatabaseError::InvalidInput);
            }
        }
        let mut seen = std::collections::BTreeSet::new();
        for row in new_rows {
            let id = row
                .get("id")
                .and_then(Value::as_i64)
                .filter(|id| *id > 0)
                .ok_or(DatabaseError::InvalidInput)?;
            if !seen.insert(id) {
                return Err(DatabaseError::InvalidInput);
            }
            let old = old_by_id.remove(&id);
            if old == Some(row) {
                continue;
            }
            if scope
                .iter()
                .filter(|(name, _)| name.as_str() != "actor_user_id")
                .any(|(name, expected)| row.get(name) != Some(expected))
            {
                return Err(DatabaseError::AccessDenied);
            }
            let fields = row.as_object().ok_or(DatabaseError::InvalidInput)?;
            let mut values = BTreeMap::new();
            for (name, value) in fields {
                if name == "id" || old.is_some_and(|old| old.get(name) == Some(value)) {
                    continue;
                }
                let field = entity
                    .fields
                    .iter()
                    .find(|field| field.name == *name)
                    .ok_or(DatabaseError::InvalidContract)?;
                let write = if let Some(target_entity) = identity_reference(entity_name, field)
                    && let Some(provisional) = value.as_i64()
                    && let Some(step) = inserted.get(&(target_entity.to_owned(), provisional))
                {
                    WriteValue::GeneratedId(*step)
                } else {
                    let value = if field.nullable
                        && field.kind == crate::operations::FieldType::Timestamp
                        && value.as_str() == Some("")
                    {
                        Value::Null
                    } else {
                        value.clone()
                    };
                    WriteValue::Literal(value)
                };
                values.insert(name.clone(), write);
            }
            if let Some(old) = old {
                let mut guard = vec![Predicate::Equal("id".to_owned(), Value::from(id))];
                for name in values.keys() {
                    let previous = old.get(name).ok_or(DatabaseError::InvalidInput)?;
                    guard.push(if previous.is_null() {
                        Predicate::IsNull(name.clone())
                    } else {
                        Predicate::Equal(name.clone(), previous.clone())
                    });
                }
                if !values.is_empty() {
                    steps.push(Mutation::Update {
                        entity: entity_name.to_owned(),
                        filter: Predicate::All(guard),
                        values,
                        expected_rows: Some(1),
                    });
                }
            } else {
                inserted.insert((entity_name.to_owned(), id), steps.len());
                steps.push(Mutation::Insert {
                    entity: entity_name.to_owned(),
                    values,
                });
            }
        }
        if !old_by_id.is_empty() {
            return Err(DatabaseError::InvalidInput);
        }
    }
    if steps.len() == 2 {
        return Ok(after.clone());
    }
    let result = adapter.transact(
        config,
        schema,
        &Transaction {
            steps,
            idempotency_key: None,
        },
    )?;
    let mut corrected = after.clone();
    for &(section, entity_name) in PAYMENT_SECTIONS {
        let entity = schema
            .entities
            .iter()
            .find(|entity| entity.name == entity_name)
            .ok_or(DatabaseError::InvalidContract)?;
        let rows = corrected
            .get_mut(section)
            .and_then(Value::as_array_mut)
            .ok_or(DatabaseError::InvalidInput)?;
        for row in rows {
            let fields = row.as_object_mut().ok_or(DatabaseError::InvalidInput)?;
            for field in &entity.fields {
                let target = if field.name == "id" {
                    Some(entity_name)
                } else {
                    identity_reference(entity_name, field)
                };
                if let Some(target) = target
                    && let Some(provisional) = fields.get(&field.name).and_then(Value::as_i64)
                    && let Some(step) = inserted.get(&(target.to_owned(), provisional))
                {
                    let actual = result
                        .generated_ids
                        .get(*step)
                        .copied()
                        .flatten()
                        .ok_or(DatabaseError::StorageFailure)?;
                    fields.insert(field.name.clone(), Value::from(actual));
                }
            }
        }
    }
    Ok(corrected)
}

fn payment_authorization_guard(
    declaration: &Value,
    actor_id: i64,
    state: &Value,
) -> Result<Mutation, DatabaseError> {
    if actor_id <= 0 {
        return Err(DatabaseError::InvalidInput);
    }
    for (entity_key, section_key, user_key, id_key) in [
        (
            "owner_entity",
            "owner_state_section",
            "owner_user_id_field",
            "owner_id_field",
        ),
        (
            "tenant_entity",
            "tenant_state_section",
            "tenant_owner_user_id_field",
            "tenant_id_field",
        ),
    ] {
        let Some(user_field) = declaration.get(user_key).and_then(Value::as_str) else {
            continue;
        };
        let entity = declared_text(declaration, entity_key)?;
        let section = declared_text(declaration, section_key)?;
        let id_field = declared_text(declaration, id_key)?;
        if let Some(row) = state
            .get(section)
            .and_then(Value::as_array)
            .and_then(|rows| rows.first())
            && row.get(user_field).and_then(Value::as_i64) == Some(actor_id)
        {
            let identity = row
                .get(id_field)
                .cloned()
                .ok_or(DatabaseError::InvalidInput)?;
            return Ok(Mutation::RequireCount {
                entity: entity.to_owned(),
                filter: Predicate::All(vec![
                    Predicate::Equal(id_field.to_owned(), identity),
                    Predicate::Equal(user_field.to_owned(), Value::from(actor_id)),
                ]),
                minimum: 1,
                maximum: Some(1),
            });
        }
    }
    let entity = declared_text(declaration, "membership_entity")?;
    let section = declared_text(declaration, "membership_state_section")?;
    let user_field = declared_text(declaration, "membership_user_id_field")?;
    let tenant_field = declared_text(declaration, "membership_tenant_id_field")?;
    let status_field = declared_text(declaration, "membership_status_field")?;
    let row = state
        .get(section)
        .and_then(Value::as_array)
        .and_then(|rows| rows.first())
        .ok_or(DatabaseError::AccessDenied)?;
    if row.get(user_field).and_then(Value::as_i64) != Some(actor_id)
        || row.get(status_field).and_then(Value::as_str) != Some("active")
    {
        return Err(DatabaseError::AccessDenied);
    }
    let mut filters = vec![
        Predicate::Equal(user_field.to_owned(), Value::from(actor_id)),
        Predicate::Equal(
            tenant_field.to_owned(),
            row.get(tenant_field)
                .cloned()
                .ok_or(DatabaseError::InvalidInput)?,
        ),
        Predicate::Equal(status_field.to_owned(), Value::String("active".to_owned())),
    ];
    if let Some(field) = declaration
        .get("membership_role_field")
        .and_then(Value::as_str)
    {
        let role = row
            .get(field)
            .and_then(Value::as_str)
            .ok_or(DatabaseError::AccessDenied)?;
        let allowed = declaration
            .get("membership_allowed_roles")
            .and_then(Value::as_array)
            .filter(|roles| !roles.is_empty() && roles.len() <= 32)
            .ok_or(DatabaseError::InvalidContract)?;
        if !allowed.iter().any(|allowed| allowed.as_str() == Some(role)) {
            return Err(DatabaseError::AccessDenied);
        }
        filters.push(Predicate::Equal(
            field.to_owned(),
            Value::String(role.to_owned()),
        ));
    }
    if let Some(field) = declaration
        .get("membership_revoked_field")
        .and_then(Value::as_str)
    {
        if row.get(field) != Some(&Value::Null) {
            return Err(DatabaseError::AccessDenied);
        }
        filters.push(Predicate::IsNull(field.to_owned()));
    }
    Ok(Mutation::RequireCount {
        entity: entity.to_owned(),
        filter: Predicate::All(filters),
        minimum: 1,
        maximum: Some(1),
    })
}