nodedb 0.3.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

// SPDX-License-Identifier: BUSL-1.1

//! Handler for `ALTER TENANT <name> IN DATABASE <db> SET QUOTA (...)`.
//!
//! Loads the tenant's stored `QuotaRecord` (or `QuotaRecord::DEFAULT`), merges
//! the partial spec, validates the result, and persists it to
//! `_system.tenant_quotas`.

use nodedb_sql::ddl_ast::AlterTenantOperation;
use nodedb_types::QuotaRecord;
use pgwire::api::results::{Response, Tag};
use pgwire::error::PgWireResult;

use crate::control::security::audit::AuditEvent;
use crate::control::security::identity::AuthenticatedIdentity;
use crate::control::state::SharedState;
use crate::types::TenantId;

use super::super::super::types::{require_tenant_admin, sqlstate_error};

/// Handle `ALTER TENANT <name> IN DATABASE <db> SET QUOTA (...)`.
pub fn handle_alter_tenant_quota(
    state: &SharedState,
    identity: &AuthenticatedIdentity,
    name: &str,
    database: &str,
    operation: &AlterTenantOperation,
) -> PgWireResult<Vec<Response>> {
    require_tenant_admin(identity, "alter tenant quota")?;

    let catalog = match state.credentials.catalog() {
        Some(c) => c,
        None => return Err(sqlstate_error("XX000", "system catalog unavailable")),
    };

    // Resolve database name → id.
    let db_id = catalog
        .get_database_id_by_name(database)
        .map_err(|e| sqlstate_error("XX000", &format!("catalog lookup failed: {e}")))?
        .ok_or_else(|| sqlstate_error("3D000", &format!("database '{database}' does not exist")))?;

    // Resolve tenant name → id via a linear scan of stored tenants.
    let tenants = catalog
        .load_all_tenants()
        .map_err(|e| sqlstate_error("XX000", &format!("tenant load failed: {e}")))?;
    let tenant_id = tenants
        .iter()
        .find(|t| t.name == name)
        .map(|t| TenantId::new(t.tenant_id))
        .ok_or_else(|| sqlstate_error("42704", &format!("tenant '{name}' does not exist")))?;

    let AlterTenantOperation::SetQuota(spec) = operation;

    // Load existing record (or DEFAULT) and keep a verbatim copy so the audit
    // entry records the exact before/after; the catalog layer enforces the
    // sum-of-tenant-quotas ≤ database-quota invariant on `put_tenant_quota`.
    let before = catalog
        .get_tenant_quota(db_id, tenant_id)
        .map_err(|e| sqlstate_error("XX000", &format!("quota read failed: {e}")))?
        .unwrap_or(QuotaRecord::DEFAULT);
    let mut record = before.clone();
    record.merge(spec);

    catalog
        .put_tenant_quota(db_id, tenant_id, &record)
        .map_err(|e| sqlstate_error("53400", &format!("{e}")))?;

    state.audit_record(
        AuditEvent::AdminAction,
        Some(tenant_id),
        &identity.username,
        &format!(
            "ALTER TENANT {name} IN DATABASE {database} SET QUOTA — before: [{}] — after: [{}]",
            before.audit_summary(),
            record.audit_summary()
        ),
    );

    Ok(vec![Response::Execution(Tag::new("ALTER TENANT"))])
}